Get each command stats from INFO all command on redis module

CHANGELOG-developer.next.asciidoc

@@ -129,6 +129,7 @@ The list below covers the major changes between 7.0.0-rc2 and master only.
 - Introduce `libbeat/beat.Beat.OutputConfigReloader` {pull}28048[28048]
 - Update Go version to 1.17.1. {pull}27543[27543]
 - Whitelist `GCP_*` environment variables in dev tools {pull}28364[28364]
+- Add `commandstats` field to Redis module {pull}29662[29662]
 
 ==== Deprecated
 

CHANGELOG.next.asciidoc

@@ -155,6 +155,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add `memory.workingset.limit.pct` field in Kubernetes container/pod metricset. {pull}29547[29547]
 - Add `elasticsearch.cluster.id` field to Beat and Kibana modules. {pull}29577[29577]
 - Add `elasticsearch.cluster.id` field to Logstash module. {pull}29625[29625]
+- Add `commandstats` field to Redis module {pull}29662[29662]
 
 *Packetbeat*
 

metricbeat/docs/fields.asciidoc

@@ -56608,6 +56608,63 @@ type: long
 Count of slow operations
 
 
+type: long
+
+--
+
+[float]
+=== commandstats
+
+Redis command statistics
+
+
+
+*`redis.info.commandstats.*.calls`*::
++
+--
+The number of calls that reached command execution (not rejected).
+
+
+type: long
+
+--
+
+*`redis.info.commandstats.*.usec`*::
++
+--
+The total CPU time consumed by these commands.
+
+
+type: long
+
+--
+
+*`redis.info.commandstats.*.usec_per_call`*::
++
+--
+The average CPU consumed per command execution.
+
+
+type: float
+
+--
+
+*`redis.info.commandstats.*.rejected_calls`*::
++
+--
+The number of rejected calls (on redis 6.2-rc2).
+
+
+type: long
+
+--
+
+*`redis.info.commandstats.*.failed_calls`*::
++
+--
+The number of failed calls (on redis 6.2-rc2).
+
+
 type: long
 
 --

metricbeat/docs/modules/redis.asciidoc

@@ -40,7 +40,7 @@ redis://HOST[:PORT][?password=PASSWORD[&db=DATABASE]]
 === Compatibility
 
 The redis metricsets `info`, `key` and `keyspace` are compatible with all distributions of Redis (OSS and enterprise).
-They were tested with Redis 3.2.12, 4.0.11 and 5.0-rc4, and are expected to work with all versions >= 3.0.
+They were tested with Redis 3.2.12, 4.0.11, 5.0-rc4 and 6.2.6, and are expected to work with all versions >= 3.0.
 
 
 [float]

metricbeat/module/redis/_meta/docs.asciidoc

@@ -31,4 +31,4 @@ redis://HOST[:PORT][?password=PASSWORD[&db=DATABASE]]
 === Compatibility
 
 The redis metricsets `info`, `key` and `keyspace` are compatible with all distributions of Redis (OSS and enterprise).
-They were tested with Redis 3.2.12, 4.0.11 and 5.0-rc4, and are expected to work with all versions >= 3.0.
+They were tested with Redis 3.2.12, 4.0.11, 5.0-rc4 and 6.2.6, and are expected to work with all versions >= 3.0.

metricbeat/module/redis/_meta/supported-versions.yml

@@ -2,3 +2,4 @@ variants:
   - REDIS_VERSION: 3.2.12
   - REDIS_VERSION: 4.0.11
   - REDIS_VERSION: 5.0.5
+  - REDIS_VERSION: 6.2.6

metricbeat/module/redis/info/_meta/data.json

@@ -19,18 +19,30 @@
         "info": {
             "clients": {
                 "blocked": 0,
-                "connected": 1,
+                "connected": 2,
                 "max_input_buffer": 0,
                 "max_output_buffer": 0
             },
             "cluster": {
                 "enabled": false
             },
+            "commandstats": {
+                "info": {
+                    "calls": 3,
+                    "usec": 241,
+                    "usec_per_call": 80.33
+                },
+                "slowlog": {
+                    "calls": 3,
+                    "usec": 22,
+                    "usec_per_call": 7.33
+                }
+            },
             "cpu": {
                 "used": {
-                    "sys": 0.23,
+                    "sys": 0.85,
                     "sys_children": 0,
-                    "user": 0.05,
+                    "user": 0.14,
                     "user_children": 0
                 }
             },
@@ -42,17 +54,17 @@
                     "rss": {}
                 },
                 "fragmentation": {
-                    "ratio": 2.85
+                    "ratio": 2.74
                 },
                 "max": {
                     "policy": "noeviction",
                     "value": 0
                 },
                 "used": {
                     "lua": 37888,
-                    "peak": 822456,
-                    "rss": 2347008,
-                    "value": 822456
+                    "peak": 843312,
+                    "rss": 2310144,
+                    "value": 843312
                 }
             },
             "persistence": {
@@ -95,7 +107,7 @@
                     "copy_on_write": {},
                     "last_save": {
                         "changes_since": 0,
-                        "time": 1633422962
+                        "time": 1642155583
                     }
                 }
             },
@@ -122,21 +134,21 @@
                 "git_dirty": "0",
                 "git_sha1": "00000000",
                 "hz": 10,
-                "lru_clock": 6033096,
+                "lru_clock": 14765805,
                 "mode": "standalone",
                 "multiplexing_api": "epoll",
-                "run_id": "ddfd6edda7dcd357ac9f24ebfb0ef8b3ee771f8d",
+                "run_id": "567ea825bae461f81d687f671c6cfaef5c1e45c7",
                 "tcp_port": 6379,
-                "uptime": 86
+                "uptime": 174
             },
             "slowlog": {
                 "count": 0
             },
             "stats": {
                 "active_defrag": {},
-                "commands_processed": 2,
+                "commands_processed": 6,
                 "connections": {
-                    "received": 82,
+                    "received": 164,
                     "rejected": 0
                 },
                 "instantaneous": {
@@ -156,10 +168,10 @@
                 "migrate_cached_sockets": 0,
                 "net": {
                     "input": {
-                        "bytes": 80
+                        "bytes": 170
                     },
                     "output": {
-                        "bytes": 2111
+                        "bytes": 6593
                     }
                 },
                 "pubsub": {
@@ -177,7 +189,7 @@
         }
     },
     "service": {
-        "address": "localhost:53854",
+        "address": "localhost:63118",
         "type": "redis",
         "version": "3.2.12"
     }

metricbeat/module/redis/info/_meta/fields.yml

@@ -552,3 +552,28 @@
       description: >
         Count of slow operations
 
+    - name: commandstats
+      type: group
+      description: >
+        Redis command statistics
+      fields:
+        - name: "*.calls"
+          type: long
+          description: >
+            The number of calls that reached command execution (not rejected).
+        - name: "*.usec"
+          type: long
+          description: >
+            The total CPU time consumed by these commands.
+        - name: "*.usec_per_call"
+          type: float
+          description: >
+            The average CPU consumed per command execution.
+        - name: "*.rejected_calls"
+          type: long
+          description: >
+            The number of rejected calls (on redis 6.2-rc2).
+        - name: "*.failed_calls"
+          type: long
+          description: >
+            The number of failed calls (on redis 6.2-rc2).

metricbeat/module/redis/info/data.go

@@ -18,6 +18,8 @@
 package info
 
 import (
+	"strings"
+
 	"github.com/elastic/beats/v7/libbeat/common"
 	s "github.com/elastic/beats/v7/libbeat/common/schema"
 	c "github.com/elastic/beats/v7/libbeat/common/schema/mapstrstr"
@@ -234,15 +236,35 @@ var (
 	}
 )
 
+func buildCommandstatsSchema(key string, schema s.Schema) {
+	// Build schema for each command
+	command := strings.Split(key, "_")[1]
+	schema[command] = s.Object{
+		"calls":          c.Int("cmdstat_" + command + "_calls"),
+		"usec":           c.Int("cmdstat_" + command + "_usec"),
+		"usec_per_call":  c.Float("cmdstat_" + command + "_usec_per_call"),
+		"rejected_calls": c.Int("cmdstat_" + command + "_rejected_calls"),
+		"failed_calls":   c.Int("cmdstat_" + command + "_failed_calls"),
+	}
+}
+
 // Map data to MapStr
 func eventMapping(r mb.ReporterV2, info map[string]string) {
 	// Full mapping from info
 	source := map[string]interface{}{}
+	commandstatsSchema := s.Schema{}
 	for key, val := range info {
 		source[key] = val
+		if strings.HasPrefix(key, "cmdstat_") {
+			buildCommandstatsSchema(key, commandstatsSchema)
+		}
 	}
 	data, _ := schema.Apply(source)
 
+	// Add commandstats info
+	commandstatsData, _ := commandstatsSchema.Apply(source)
+	data["commandstats"] = commandstatsData
+
 	rootFields := common.MapStr{}
 	if v, err := data.GetValue("server.version"); err == nil {
 		rootFields.Put("service.version", v)

metricbeat/module/redis/info/info.go

@@ -59,8 +59,8 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 		}
 	}()
 
-	// Fetch default INFO.
-	info, err := redis.FetchRedisInfo("default", conn)
+	// Fetch all INFO.
+	info, err := redis.FetchRedisInfo("all", conn)
 	if err != nil {
 		return errors.Wrap(err, "failed to fetch redis info")
 	}

metricbeat/module/redis/info/info_integration_test.go

@@ -46,7 +46,7 @@ func TestFetch(t *testing.T) {
 	t.Logf("%s/%s event: %+v", ms.Module().Name(), ms.Name(), event)
 
 	// Check fields
-	assert.Equal(t, 9, len(event))
+	assert.Equal(t, 10, len(event))
 	server := event["server"].(common.MapStr)
 	assert.Equal(t, "standalone", server["mode"])
 }

metricbeat/module/redis/redis.go

@@ -52,7 +52,15 @@ func ParseRedisInfo(info string) map[string]string {
 		// Values are separated by :
 		parts := ParseRedisLine(value, ":")
 		if len(parts) == 2 {
-			values[parts[0]] = parts[1]
+			if strings.Contains(parts[0], "cmdstat_") {
+				cmdstats := ParseRedisCommandStats(parts[0], parts[1])
+				for k, v := range cmdstats {
+					key := parts[0] + "_" + k
+					values[key] = v
+				}
+			} else {
+				values[parts[0]] = parts[1]
+			}
 		}
 	}
 	return values
@@ -63,6 +71,22 @@ func ParseRedisLine(s string, delimiter string) []string {
 	return strings.Split(s, delimiter)
 }
 
+// ParseRedisCommandStats parses a map of stats returned by INFO COMMANDSTATS
+func ParseRedisCommandStats(key string, s string) map[string]string {
+	// calls=XX,usec=XXX,usec_per_call=XXX
+	results := strings.Split(s, ",")
+
+	values := map[string]string{}
+
+	for _, value := range results {
+		parts := strings.Split(value, "=")
+		if len(parts) == 2 {
+			values[parts[0]] = parts[1]
+		}
+	}
+	return values
+}
+
 // FetchRedisInfo returns a map of requested stats.
 func FetchRedisInfo(stat string, c rd.Conn) (map[string]string, error) {
 	out, err := rd.String(c.Do("INFO", stat))

metricbeat/module/redis/test_redis.py

@@ -8,7 +8,7 @@
 
 REDIS_FIELDS = metricbeat.COMMON_FIELDS + ["redis"]
 
-REDIS_INFO_FIELDS = ["clients", "cluster", "cpu", "memory",
+REDIS_INFO_FIELDS = ["clients", "cluster", "commandstats", "cpu", "memory",
                      "persistence", "replication", "server", "stats", "slowlog"]
 
 REDIS_KEYSPACE_FIELDS = ["keys", "expires", "id", "avg_ttl"]