Rabbitmq queue metricset message rates

CHANGELOG.asciidoc

@@ -121,7 +121,6 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 
 *Metricbeat*
 
-- Add connections metricset to RabbitMQ module {pull}6548[6548]
 - Change field type of http header from nested to object {pull}5258[5258]
 - Fix the fetching of process information when some data is missing under MacOS X. {issue}5337[5337]
 - Change `MySQL active connections` visualization title to `MySQL total connections`. {issue}4812[4812]
@@ -256,6 +255,8 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Making the MongoDB module GA. {pull}6554[6554]
 - Allow to disable labels `dedot` in Docker module, in favor of a safe way to keep dots. {pull}6490[6490]
 - Add experimental module to collect metrics from munin nodes. {pull}6517[6517]
+- Add connections metricset to RabbitMQ module {pull}6548[6548]
+- Add message rates to the RabbitMQ queue metricset {issue}6442[6442] {pull}6606[6606]
 
 *Packetbeat*
 

libbeat/common/schema/mapstriface/mapstriface.go

@@ -171,6 +171,11 @@ func Bool(key string, opts ...schema.SchemaOption) schema.Conv {
 	return schema.SetOptions(schema.Conv{Key: key, Func: toBool}, opts)
 }
 
+// Float creates a Conv object for parsing floats
+func Float(key string, opts ...schema.SchemaOption) schema.Conv {
+	return schema.SetOptions(schema.Conv{Key: key, Func: toInteger}, opts)
+}
+
 func toInteger(key string, data map[string]interface{}) (interface{}, error) {
 	emptyIface, exists := data[key]
 	if !exists {

metricbeat/docs/fields.asciidoc

@@ -9230,6 +9230,14 @@ type: long
 Sum of ready and unacknowledged messages (queue depth).
 
 
+[float]
+=== `rabbitmq.queue.messages.total.details.rate`
+
+type: float
+
+How much the queue depth has changed per second in the most recent sampling interval.
+
+
 [float]
 === `rabbitmq.queue.messages.ready.count`
 
@@ -9238,6 +9246,14 @@ type: long
 Number of messages ready to be delivered to clients.
 
 
+[float]
+=== `rabbitmq.queue.messages.ready.details.rate`
+
+type: float
+
+How much the count of messages ready has changed per second in the most recent sampling interval.
+
+
 [float]
 === `rabbitmq.queue.messages.unacknowledged.count`
 
@@ -9246,6 +9262,14 @@ type: long
 Number of messages delivered to clients but not yet acknowledged.
 
 
+[float]
+=== `rabbitmq.queue.messages.unacknowledged.details.rate`
+
+type: float
+
+How much the count of unacknowledged messages has changed per second in the most recent sampling interval.
+
+
 [float]
 === `rabbitmq.queue.messages.persistent.count`
 

metricbeat/module/rabbitmq/_meta/testdata/queue_sample_response.json

@@ -65,15 +65,15 @@
         },
         "messages": 74,
         "messages_details": {
-            "rate": 0
+            "rate": 2.2
         },
         "messages_ready": 71,
         "messages_ready_details": {
             "rate": 0
         },
         "messages_unacknowledged": 3,
         "messages_unacknowledged_details": {
-            "rate": 0
+            "rate": 0.5
         },
         "idle_since": "2017-07-28 23:45:52",
         "consumer_utilisation": 0.7,

metricbeat/module/rabbitmq/queue/_meta/data.json

@@ -25,13 +25,22 @@
         "state": "running",
         "messages": {
           "total": {
-            "count": 0
+            "count": 0,
+            "details": {
+              "rate": 2.2
+            }
           },
           "ready": {
-            "count": 0
+            "count": 0,
+            "details": {
+              "rate": 0
+            }
           },
           "unacknowledged": {
-            "count": 0
+            "count": 0,
+            "details": {
+              "rate": 0.5
+            }
           },
           "persistent": {
             "count": 0

metricbeat/module/rabbitmq/queue/_meta/fields.yml

@@ -49,14 +49,26 @@
       type: long
       description: >
         Sum of ready and unacknowledged messages (queue depth).
+    - name: messages.total.details.rate
+      type: float
+      description: >
+        How much the queue depth has changed per second in the most recent sampling interval.
     - name: messages.ready.count
       type: long
       description: >
         Number of messages ready to be delivered to clients.
+    - name: messages.ready.details.rate
+      type: float
+      description: >
+        How much the count of messages ready has changed per second in the most recent sampling interval.
     - name: messages.unacknowledged.count
       type: long
       description: >
         Number of messages delivered to clients but not yet acknowledged.
+    - name: messages.unacknowledged.details.rate
+      type: float
+      description: >
+        How much the count of unacknowledged messages has changed per second in the most recent sampling interval.
     - name: messages.persistent.count
       type: long
       description: >

metricbeat/module/rabbitmq/queue/data.go

@@ -30,12 +30,21 @@ var (
 		"messages": s.Object{
 			"total": s.Object{
 				"count": c.Int("messages"),
+				"details": c.Dict("messages_details", s.Schema{
+					"rate": c.Float("rate"),
+				}),
 			},
 			"ready": s.Object{
 				"count": c.Int("messages_ready"),
+				"details": c.Dict("messages_ready_details", s.Schema{
+					"rate": c.Float("rate"),
+				}),
 			},
 			"unacknowledged": s.Object{
 				"count": c.Int("messages_unacknowledged"),
+				"details": c.Dict("messages_unacknowledged_details", s.Schema{
+					"rate": c.Float("rate"),
+				}),
 			},
 			"persistent": s.Object{
 				"count": c.Int("messages_persistent"),

metricbeat/module/rabbitmq/queue/node_test.go → metricbeat/module/rabbitmq/queue/queue_test.go

@@ -68,6 +68,15 @@ func TestFetchEventContents(t *testing.T) {
 	assert.EqualValues(t, 3, unacknowledged["count"])
 	assert.EqualValues(t, 73, persistent["count"])
 
+	totalDetails := total["details"].(common.MapStr)
+	assert.EqualValues(t, 2.2, totalDetails["rate"])
+
+	readyDetails := ready["details"].(common.MapStr)
+	assert.EqualValues(t, 0, readyDetails["rate"])
+
+	unacknowledgedDetails := unacknowledged["details"].(common.MapStr)
+	assert.EqualValues(t, 0.5, unacknowledgedDetails["rate"])
+
 	disk := event["disk"].(common.MapStr)
 	reads := disk["reads"].(common.MapStr)
 	writes := disk["writes"].(common.MapStr)