Rabbitmq exchange metricset and queue message rates (#6442)

CHANGELOG.asciidoc

@@ -115,7 +115,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]
@@ -235,6 +234,8 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Refactor prometheus endpoint parsing to look similar to upstream prometheus {pull}6332[6332]
 - Update prometheus dependencies to latest {pull}6333[6333]
 - Making the http/json metricset GA. {pull}6471[6471]
+- Add connections metricset to RabbitMQ module {pull}6548[6548]
+- Add exchanges metricset and queue message rates to RabbitMQ module {issue}6442[6442] {pull}6586[6586]
 
 *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

@@ -8745,6 +8745,181 @@ type: long
 Number of octets received on the connection.
 
 
+[float]
+== exchange fields
+
+exchange
+
+
+
+[float]
+=== `rabbitmq.exchange.name`
+
+type: keyword
+
+The name of the queue with non-ASCII characters escaped as in C.
+
+
+[float]
+=== `rabbitmq.exchange.vhost`
+
+type: keyword
+
+Virtual host name with non-ASCII characters escaped as in C.
+
+
+[float]
+=== `rabbitmq.exchange.durable`
+
+type: boolean
+
+Whether or not the queue survives server restarts.
+
+
+[float]
+=== `rabbitmq.exchange.auto_delete`
+
+type: boolean
+
+Whether the queue will be deleted automatically when no longer used.
+
+
+[float]
+=== `rabbitmq.exchange.internal`
+
+type: boolean
+
+Whether the exchange is internal, i.e. cannot be directly published to by a client.
+
+
+[float]
+=== `rabbitmq.exchange.messages.publish.count`
+
+type: long
+
+Count of messages published.
+
+
+[float]
+=== `rabbitmq.exchange.messages.publish.details.rate`
+
+type: long
+
+How much the exchange publish count has changed per second in the most recent sampling interval.
+
+
+[float]
+=== `rabbitmq.exchange.messages.publish_in.count`
+
+type: long
+
+Count of messages published "in" to an exchange, i.e. not taking account of routing.
+
+
+[float]
+=== `rabbitmq.exchange.messages.publish_in.details.rate`
+
+type: long
+
+How much the exchange publish-in count has changed per second in the most recent sampling interval.
+
+
+[float]
+=== `rabbitmq.exchange.messages.publish_out.count`
+
+type: long
+
+Count of messages published "out" of an exchange, i.e. taking account of routing.
+
+
+[float]
+=== `rabbitmq.exchange.messages.publish_out.details.rate`
+
+type: long
+
+How much the exchange publish-out count has changed per second in the most recent sampling interval.
+
+
+[float]
+=== `rabbitmq.exchange.messages.ack.count`
+
+type: long
+
+Count of acknowledged messages.
+
+
+[float]
+=== `rabbitmq.exchange.messages.ack.details.rate`
+
+type: long
+
+How much the exchange ack count has changed per second in the most recent sampling interval.
+
+
+[float]
+=== `rabbitmq.exchange.messages.deliver_get.count`
+
+type: long
+
+Count of all messages delivered to consumers.
+
+
+[float]
+=== `rabbitmq.exchange.messages.deliver_get.details.rate`
+
+type: long
+
+How much the exchange deliver-get count has changed per second in the most recent sampling interval.
+
+
+[float]
+=== `rabbitmq.exchange.messages.confirm.count`
+
+type: long
+
+Count of messages confirmed.
+
+
+[float]
+=== `rabbitmq.exchange.messages.confirm.details.rate`
+
+type: long
+
+How much the exchange confirm count has changed per second in the most recent sampling interval.
+
+
+[float]
+=== `rabbitmq.exchange.messages.return_unroutable.count`
+
+type: long
+
+Count of messages returned to publisher as unroutable.
+
+
+[float]
+=== `rabbitmq.exchange.messages.return_unroutable.details.rate`
+
+type: long
+
+How much the exchange return-unroutable count has changed per second in the most recent sampling interval.
+
+
+[float]
+=== `rabbitmq.exchange.messages.redeliver.count`
+
+type: long
+
+Count of subset of messages in deliver_get which had the redelivered flag set.
+
+
+[float]
+=== `rabbitmq.exchange.messages.redeliver.details.rate`
+
+type: long
+
+How much the exchange redeliver count has changed per second in the most recent sampling interval.
+
+
 [float]
 == node fields
 
@@ -9157,6 +9332,14 @@ type: long
 Sum of ready and unacknowledged messages (queue depth).
 
 
+[float]
+=== `rabbitmq.queue.messages.total.details.rate`
+
+type: long
+
+How much the queue depth has changed per second in the most recent sampling interval.
+
+
 [float]
 === `rabbitmq.queue.messages.ready.count`
 
@@ -9165,6 +9348,14 @@ type: long
 Number of messages ready to be delivered to clients.
 
 
+[float]
+=== `rabbitmq.queue.messages.ready.details.rate`
+
+type: long
+
+How much the count of messages ready has changed per second in the most recent sampling interval.
+
+
 [float]
 === `rabbitmq.queue.messages.unacknowledged.count`
 
@@ -9173,6 +9364,14 @@ type: long
 Number of messages delivered to clients but not yet acknowledged.
 
 
+[float]
+=== `rabbitmq.queue.messages.unacknowledged.details.rate`
+
+type: long
+
+How much the count of unacknowledged messages has changed per second in the most recent sampling interval.
+
+
 [float]
 === `rabbitmq.queue.messages.persistent.count`
 

metricbeat/docs/modules/rabbitmq.asciidoc

@@ -21,7 +21,7 @@ in <<configuration-metricbeat>>. Here is an example configuration:
 ----
 metricbeat.modules:
 - module: rabbitmq
-  metricsets: ["node", "queue", "connection"]
+  metricsets: ["node", "queue", "connection", "exchange"]
   period: 10s
   hosts: ["localhost:15672"]
 
@@ -38,12 +38,16 @@ The following metricsets are available:
 
 * <<metricbeat-metricset-rabbitmq-connection,connection>>
 
+* <<metricbeat-metricset-rabbitmq-exchange,exchange>>
+
 * <<metricbeat-metricset-rabbitmq-node,node>>
 
 * <<metricbeat-metricset-rabbitmq-queue,queue>>
 
 include::rabbitmq/connection.asciidoc[]
 
+include::rabbitmq/exchange.asciidoc[]
+
 include::rabbitmq/node.asciidoc[]
 
 include::rabbitmq/queue.asciidoc[]

metricbeat/docs/modules/rabbitmq/exchange.asciidoc

@@ -0,0 +1,23 @@
+////
+This file is generated! See scripts/docs_collector.py
+////
+
+[[metricbeat-metricset-rabbitmq-exchange]]
+=== RabbitMQ exchange metricset
+
+beta[]
+
+include::../../../module/rabbitmq/exchange/_meta/docs.asciidoc[]
+
+
+==== Fields
+
+For a description of each field in the metricset, see the
+<<exported-fields-rabbitmq,exported fields>> section.
+
+Here is an example document generated by this metricset:
+
+[source,json]
+----
+include::../../../module/rabbitmq/exchange/_meta/data.json[]
+----

metricbeat/docs/modules_list.asciidoc

@@ -93,7 +93,8 @@ This file is generated! See scripts/docs_collector.py
 .2+|   |<<metricbeat-metricset-prometheus-collector,collector>> beta[]  
 |<<metricbeat-metricset-prometheus-stats,stats>> beta[]  
 |<<metricbeat-module-rabbitmq,RabbitMQ>>  beta[]   |    
-.3+|   |<<metricbeat-metricset-rabbitmq-connection,connection>> beta[]  
+.4+|   |<<metricbeat-metricset-rabbitmq-connection,connection>> beta[]  
+|<<metricbeat-metricset-rabbitmq-exchange,exchange>> beta[]  
 |<<metricbeat-metricset-rabbitmq-node,node>> beta[]  
 |<<metricbeat-metricset-rabbitmq-queue,queue>> beta[]  
 |<<metricbeat-module-redis,Redis>>     |    

metricbeat/include/list.go

@@ -98,6 +98,7 @@ import (
 	_ "github.com/elastic/beats/metricbeat/module/prometheus/stats"
 	_ "github.com/elastic/beats/metricbeat/module/rabbitmq"
 	_ "github.com/elastic/beats/metricbeat/module/rabbitmq/connection"
+	_ "github.com/elastic/beats/metricbeat/module/rabbitmq/exchange"
 	_ "github.com/elastic/beats/metricbeat/module/rabbitmq/node"
 	_ "github.com/elastic/beats/metricbeat/module/rabbitmq/queue"
 	_ "github.com/elastic/beats/metricbeat/module/redis"

metricbeat/metricbeat.reference.yml

@@ -416,7 +416,7 @@ metricbeat.modules:
 
 #------------------------------ RabbitMQ Module ------------------------------
 - module: rabbitmq
-  metricsets: ["node", "queue", "connection"]
+  metricsets: ["node", "queue", "connection", "exchange"]
   period: 10s
   hosts: ["localhost:15672"]
 

metricbeat/module/rabbitmq/_meta/config.yml

@@ -1,5 +1,5 @@
 - module: rabbitmq
-  metricsets: ["node", "queue", "connection"]
+  metricsets: ["node", "queue", "connection", "exchange"]
   period: 10s
   hosts: ["localhost:15672"]
 

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

@@ -0,0 +1,45 @@
+[
+    {
+        "message_stats": {
+            "publish": 123,
+            "publish_details": {
+                "rate": 0.1
+            },
+            "publish_in": 100,
+            "publish_in_details": {
+                "rate": 0.5
+            },
+            "publish_out": 99,
+            "publish_out_details": {
+                "rate": 0.9
+            },
+            "ack": 60,
+            "ack_details": {
+                "rate": 12.5
+            },
+            "deliver_get": 50,
+            "deliver_get_details": {
+                "rate": 43.21
+            },
+            "confirm": 120,
+            "confirm_details": {
+                "rate": 98.63
+            },
+            "return_unroutable": 40,
+            "return_unroutable_details": {
+                "rate": 123
+            },
+            "redeliver": 30,
+            "redeliver_details": {
+                "rate": 0
+            }
+        },
+        "name": "exchange.name",
+        "vhost": "/",
+        "type": "fanout",
+        "durable": true,
+        "auto_delete": false,
+        "internal": false,
+        "arguments": {}
+    }
+]