Metricbeat exchanges and queues' message rates (enhancement request)

[spotlesscoder]

From https://discuss.elastic.co/t/metricbeat-rabbitmq-collect-message-rates/120910

It would be great to also have the incoming/outgoing message rates fields collected by metricbeat for both queues and exchanges.

This way, I could monitor whether someone is still sending data over my RMQ instance.

[spotlesscoder]

I try to figure out how the schema works.

I found these lines in the metricbeat/module/rabbitmq/queue/data.go file:

var (
	schema = s.Schema{
		"name":        c.Str("name"),
		"vhost":       c.Str("vhost"),
		"durable":     c.Bool("durable"),
		"auto_delete": c.Bool("auto_delete"),
		"exclusive":   c.Bool("exclusive"),
		"node":        c.Str("node"),
		"state":       c.Str("state"),
		"arguments": c.Dict("arguments", s.Schema{
			"max_priority": c.Int("x-max-priority", s.Optional),
		}),
		"consumers": s.Object{
			"count": c.Int("consumers"),
			"utilisation": s.Object{
				"pct": c.Int("consumer_utilisation", s.Optional),
			},
		},
		"messages": s.Object{
			"total": s.Object{
				"count": c.Int("messages"),
			},
			"ready": s.Object{
				"count": c.Int("messages_ready"),
			},
			"unacknowledged": s.Object{
				"count": c.Int("messages_unacknowledged"),
			},
			"persistent": s.Object{
				"count": c.Int("messages_persistent"),
			},
		},
		"memory": s.Object{
			"bytes": c.Int("memory"),
		},
		"disk": s.Object{
			"reads": s.Object{
				"count": c.Int("disk_reads"),
			},
			"writes": s.Object{
				"count": c.Int("disk_writes"),
			},
		},
	}
)

I'm not sure how to get the data. In the result from the rabbitmq API, the "rate" keys do not have unique names for all of the 3 occurences:

{"memory":10000,"messages":0,"messages_details":{"rate":0.0},"messages_ready":0,"messages_ready_details":{"rate":0.0},"messages_unacknowledged":0,"messages_unacknowledged_details":{"rate":0.0},"idle_since":"2018-03-06 20:45:20","consumer_utilisation":null,"policy":null,"exclusive_consumer_tag":null,"consumers":0,"recoverable_slaves":null,"state":"running","messages_ram":0,"messages_ready_ram":0,"messages_unacknowledged_ram":0,"messages_persistent":0,"message_bytes":0,"message_bytes_ready":0,"message_bytes_unacknowledged":0,"message_bytes_ram":0,"message_bytes_persistent":0,"disk_reads":0,"disk_writes":0,"backing_queue_status":{"q1":0,"q2":0,"delta":["delta","undefined",0,"undefined"],"q3":0,"q4":0,"len":0,"target_ram_count":"infinity","next_seq_id":0,"avg_ingress_rate":0.0,"avg_egress_rate":0.0,"avg_ack_ingress_rate":0.0,"avg_ack_egress_rate":0.0},"incoming":[],"deliveries":[],"consumer_details":[],"name":"test","vhost":"test","durable":true,"auto_delete":false,"arguments":{},"node":"rabbit@host"}

I'm not sure whether the following structure would work. Does the "messages_details" key uniquely identify the underlying "rate" key in the source dataset? Is this even possible with the source data at the moment or would it be better to have unique keys for all of the 3 "rate" keys?

"messages_details": s.Object{
	"rate_incoming:": c.Str("rate"),
},

Also, I didn't find out how to get the float/double value using the existing source code, so I used string for now. How can I use float / double. Typing "c." does not suggest anything like that in autocomplete.

As you can see, I'm completely new to the go programming language and I'd be very happy if anyone could help me.

[spotlesscoder]

@kvalev could you give me a hint maybe? I looked up who created the file and found your name ;)

[kvalev]

Do you know whether you can get the message rates from the RabbitMQ REST API? Looking at the REST API output of my RabbitMQ version there seems to be backing_queue_status.avg_ingress_rate/avg_egress_rate, but I am not sure whether those are the ones that you need.
Declaring the type should be easy - you can do c.Float. I dont know about autocomplete - my editor does not support it as well, so I just looked for some examples in the existing codebase. Here is one for float -

beats/metricbeat/module/apache/status/data.go

Line 22 in 2909400

"requests_per_sec": c.Float("ReqPerSec", s.Optional),

[spotlesscoder]

The REST API has the following endpoint: /api/queues/vhost/name

Which gives me the following result for my example queue

{"consumer_details":[],"incoming":[],"deliveries":[],"messages_details":{"rate":4.6},"messages":68,"messages_unacknowledged_details":{"rate":0.0},"messages_unacknowledged":0,"messages_ready_details":{"rate":4.6},"messages_ready":68,"reductions_details":{"rate":2314.6},"reductions":40333,"message_stats":{"publish_details":{"rate":1.4},"publish":68},"node":"rabbit@rabbit-1","arguments":{},"exclusive":false,"auto_delete":false,"durable":true,"vhost":"vh","name":"test","message_bytes_paged_out":0,"messages_paged_out":0,"backing_queue_status":{"avg_ack_egress_rate":0.0,"avg_ack_ingress_rate":0.0,"avg_egress_rate":0.0,"avg_ingress_rate":0.5595230189557673,"delta":["delta","undefined",0,0,"undefined"],"len":68,"mode":"default","next_seq_id":68,"q1":0,"q2":0,"q3":0,"q4":68,"target_ram_count":"infinity"},"head_message_timestamp":null,"message_bytes_persistent":0,"message_bytes_ram":1026,"message_bytes_unacknowledged":0,"message_bytes_ready":1026,"message_bytes":1026,"messages_persistent":0,"messages_unacknowledged_ram":0,"messages_ready_ram":68,"messages_ram":68,"garbage_collection":{"minor_gcs":26,"fullsweep_after":65535,"min_heap_size":233,"min_bin_vheap_size":46422,"max_heap_size":0},"state":"running","recoverable_slaves":null,"consumers":0,"exclusive_consumer_tag":null,"effective_policy_definition":[],"operator_policy":null,"policy":null,"consumer_utilisation":null,"idle_since":"2018-03-09 17:01:08","memory":54312}

I think the data I need is in the following keys: "messages_details":{"rate":4.6},"messages":68,"messages_unacknowledged_details":{"rate":0.0},"messages_unacknowledged":0,"messages_ready_details":{"rate":4.6}

But the fields are all called "rate" so I wonder how I can access the correct one

[kvalev]

I think somebody on the forums had the same issue - https://discuss.elastic.co/t/libbeat-common-schema-mapstriface/118678

[spotlesscoder]

I still don't really understand how to solve it.

What I understood from the last post of @urso (https://discuss.elastic.co/t/libbeat-common-schema-mapstriface/118678/6?u=christian_h) is that I need to use mapstriface rather than mapstrstr

But I still don't understand which identifier I need to access the different "rate" fields.
Maybe like this? (see line 22)

var (
	schema = s.Schema{
		"name":        c.Str("name"),
		"vhost":       c.Str("vhost"),
		"durable":     c.Bool("durable"),
		"auto_delete": c.Bool("auto_delete"),
		"exclusive":   c.Bool("exclusive"),
		"node":        c.Str("node"),
		"state":       c.Str("state"),
		"arguments": c.Dict("arguments", s.Schema{
			"max_priority": c.Int("x-max-priority", s.Optional),
		}),
		"consumers": s.Object{
			"count": c.Int("consumers"),
			"utilisation": s.Object{
				"pct": c.Int("consumer_utilisation", s.Optional),
			},
		},
		"messages": s.Object{
			"total": s.Object{
				"count": c.Int("messages"),
                                "rate": c.Float("messages_details.rate"),
			},
			"ready": s.Object{
				"count": c.Int("messages_ready"),
			},
			"unacknowledged": s.Object{
				"count": c.Int("messages_unacknowledged"),
			},
			"persistent": s.Object{
				"count": c.Int("messages_persistent"),
			},
		},
		"memory": s.Object{
			"bytes": c.Int("memory"),
		},
		"disk": s.Object{
			"reads": s.Object{
				"count": c.Int("disk_reads"),
			},
			"writes": s.Object{
				"count": c.Int("disk_writes"),
			},
		},
	}
)

[kvalev]

I have started working on pull request, which I can submit in few days - master...kvalev:rabbitmq_message_rates

[spotlesscoder]

Great, thanks!

[ruflin]

@CodingSpiderFox @kvalev Thanks for working on this. I'm looking forward to the PR this will also make commenting easier. No worries if the PR is still WIP and does not fully work yet.

[spotlesscoder]

Is there anything left which needs to be done to be able to create the PR?