telegraf/plugins/rabbitmq/rabbitmq.go

337 lines
8.3 KiB
Go
Raw Normal View History

2015-07-21 17:29:25 +00:00
package rabbitmq
import (
"encoding/json"
"fmt"
"net/http"
"strconv"
2015-07-21 17:29:25 +00:00
"github.com/influxdb/telegraf/plugins"
)
const DefaultUsername = "guest"
const DefaultPassword = "guest"
const DefaultURL = "http://localhost:15672"
type Server struct {
URL string
Name string
2015-07-21 17:29:25 +00:00
Username string
Password string
Nodes []string
Queues []string
2015-07-21 17:29:25 +00:00
}
type RabbitMQ struct {
Servers []*Server
Client *http.Client
}
type OverviewResponse struct {
MessageStats *MessageStats `json:"message_stats"`
ObjectTotals *ObjectTotals `json:"object_totals"`
QueueTotals *QueueTotals `json:"queue_totals"`
}
type Details struct {
Rate float64
}
2015-07-21 17:29:25 +00:00
type MessageStats struct {
Ack int64
AckDetails Details `json:"ack_details"`
Deliver int64
DeliverDetails Details `json:"deliver_details"`
DeliverGet int64
DeliverGetDetails Details `json:"deliver_get_details"`
Publish int64
PublishDetails Details `json:"publish_details"`
Redeliver int64
RedeliverDetails Details `json:"redeliver_details"`
2015-07-21 17:29:25 +00:00
}
type ObjectTotals struct {
Channels int64
Connections int64
Consumers int64
Exchanges int64
Queues int64
}
type QueueTotals struct {
Messages int64
MessagesReady int64 `json:"messages_ready"`
MessagesUnacknowledged int64 `json:"messages_unacknowledged"`
}
type Queue struct {
QueueTotals // just to not repeat the same code
MessageStats `json:"message_stats"`
Memory int64
Consumers int64
ConsumerUtilisation float64 `json:"consumer_utilisation"`
Name string
Node string
Vhost string
Durable bool
AutoDelete bool `json:"auto_delete"`
}
2015-07-21 17:29:25 +00:00
type Node struct {
Name string
DiskFree int64 `json:"disk_free"`
DiskFreeLimit int64 `json:"disk_free_limit"`
FdTotal int64 `json:"fd_total"`
FdUsed int64 `json:"fd_used"`
MemLimit int64 `json:"mem_limit"`
MemUsed int64 `json:"mem_used"`
ProcTotal int64 `json:"proc_total"`
ProcUsed int64 `json:"proc_used"`
RunQueue int64 `json:"run_queue"`
SocketsTotal int64 `json:"sockets_total"`
SocketsUsed int64 `json:"sockets_used"`
}
type gatherFunc func(r *RabbitMQ, serv *Server, acc plugins.Accumulator, errChan chan error)
var gatherFunctions = []gatherFunc{gatherOverview, gatherNodes, gatherQueues}
2015-07-21 17:29:25 +00:00
var sampleConfig = `
# Specify servers via an array of tables
[[rabbitmq.servers]]
# name = "rmq-server-1" # optional tag
# url = "http://localhost:15672"
# username = "guest"
# password = "guest"
# A list of nodes to pull metrics about. If not specified, metrics for
# all nodes are gathered.
# nodes = ["rabbit@node1", "rabbit@node2"]
2015-07-21 17:29:25 +00:00
`
func (r *RabbitMQ) SampleConfig() string {
return sampleConfig
}
func (r *RabbitMQ) Description() string {
return "Read metrics from one or many RabbitMQ servers via the management API"
}
var localhost = &Server{URL: DefaultURL}
func (r *RabbitMQ) Gather(acc plugins.Accumulator) error {
if r.Client == nil {
r.Client = &http.Client{}
}
var errChan = make(chan error, len(r.Servers))
// use localhost is no servers are specified in config
2015-07-21 17:29:25 +00:00
if len(r.Servers) == 0 {
r.Servers = append(r.Servers, localhost)
2015-07-21 17:29:25 +00:00
}
for _, serv := range r.Servers {
for _, f := range gatherFunctions {
go f(r, serv, acc, errChan)
}
}
for i := 1; i <= len(r.Servers)*len(gatherFunctions); i++ {
err := <-errChan
2015-07-21 17:29:25 +00:00
if err != nil {
return err
}
}
return nil
}
func (r *RabbitMQ) requestJSON(serv *Server, u string, target interface{}) error {
u = fmt.Sprintf("%s%s", serv.URL, u)
req, err := http.NewRequest("GET", u, nil)
if err != nil {
return err
}
username := serv.Username
if username == "" {
username = DefaultUsername
}
password := serv.Password
if password == "" {
password = DefaultPassword
}
req.SetBasicAuth(username, password)
resp, err := r.Client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
json.NewDecoder(resp.Body).Decode(target)
return nil
}
func gatherOverview(r *RabbitMQ, serv *Server, acc plugins.Accumulator, errChan chan error) {
2015-07-21 17:29:25 +00:00
overview := &OverviewResponse{}
err := r.requestJSON(serv, "/api/overview", &overview)
if err != nil {
errChan <- err
return
2015-07-21 17:29:25 +00:00
}
if overview.QueueTotals == nil || overview.ObjectTotals == nil || overview.MessageStats == nil {
errChan <- fmt.Errorf("Wrong answer from rabbitmq. Probably auth issue")
return
}
tags := map[string]string{"url": serv.URL}
if serv.Name != "" {
tags["name"] = serv.Name
}
2015-07-21 17:29:25 +00:00
acc.Add("messages", overview.QueueTotals.Messages, tags)
acc.Add("messages_ready", overview.QueueTotals.MessagesReady, tags)
acc.Add("messages_unacked", overview.QueueTotals.MessagesUnacknowledged, tags)
acc.Add("channels", overview.ObjectTotals.Channels, tags)
acc.Add("connections", overview.ObjectTotals.Connections, tags)
acc.Add("consumers", overview.ObjectTotals.Consumers, tags)
acc.Add("exchanges", overview.ObjectTotals.Exchanges, tags)
acc.Add("queues", overview.ObjectTotals.Queues, tags)
acc.Add("messages_acked", overview.MessageStats.Ack, tags)
acc.Add("messages_delivered", overview.MessageStats.Deliver, tags)
acc.Add("messages_published", overview.MessageStats.Publish, tags)
2015-07-21 17:29:25 +00:00
errChan <- nil
}
2015-07-21 17:29:25 +00:00
func gatherNodes(r *RabbitMQ, serv *Server, acc plugins.Accumulator, errChan chan error) {
nodes := make([]Node, 0)
// Gather information about nodes
err := r.requestJSON(serv, "/api/nodes", &nodes)
2015-07-21 17:29:25 +00:00
if err != nil {
errChan <- err
return
2015-07-21 17:29:25 +00:00
}
for _, node := range nodes {
if !shouldGatherNode(node, serv) {
continue
}
tags := map[string]string{"url": serv.URL}
tags["node"] = node.Name
2015-07-21 17:29:25 +00:00
acc.Add("disk_free", node.DiskFree, tags)
acc.Add("disk_free_limit", node.DiskFreeLimit, tags)
acc.Add("fd_total", node.FdTotal, tags)
acc.Add("fd_used", node.FdUsed, tags)
acc.Add("mem_limit", node.MemLimit, tags)
acc.Add("mem_used", node.MemUsed, tags)
acc.Add("proc_total", node.ProcTotal, tags)
acc.Add("proc_used", node.ProcUsed, tags)
acc.Add("run_queue", node.RunQueue, tags)
acc.Add("sockets_total", node.SocketsTotal, tags)
acc.Add("sockets_used", node.SocketsUsed, tags)
}
errChan <- nil
}
func gatherQueues(r *RabbitMQ, serv *Server, acc plugins.Accumulator, errChan chan error) {
// Gather information about queues
queues := make([]Queue, 0)
err := r.requestJSON(serv, "/api/queues", &queues)
if err != nil {
errChan <- err
return
}
for _, queue := range queues {
if !shouldGatherQueue(queue, serv) {
continue
}
tags := map[string]string{
"url": serv.URL,
"queue": queue.Name,
"vhost": queue.Vhost,
"node": queue.Node,
"durable": strconv.FormatBool(queue.Durable),
"auto_delete": strconv.FormatBool(queue.AutoDelete),
}
acc.AddFields(
"queue",
map[string]interface{}{
// common information
"consumers": queue.Consumers,
"consumer_utilisation": queue.ConsumerUtilisation,
"memory": queue.Memory,
// messages information
"messages": queue.Messages,
"messages_ready": queue.MessagesReady,
"messages_unack": queue.MessagesUnacknowledged,
"messages_ack": queue.MessageStats.Ack,
"messages_ack_rate": queue.MessageStats.AckDetails.Rate,
"messages_deliver": queue.MessageStats.Deliver,
"messages_deliver_rate": queue.MessageStats.DeliverDetails.Rate,
"messages_deliver_get": queue.MessageStats.DeliverGet,
"messages_deliver_get_rate": queue.MessageStats.DeliverGetDetails.Rate,
"messages_publish": queue.MessageStats.Publish,
"messages_publish_rate": queue.MessageStats.PublishDetails.Rate,
"messages_redeliver": queue.MessageStats.Redeliver,
"messages_redeliver_rate": queue.MessageStats.RedeliverDetails.Rate,
},
tags,
)
}
errChan <- nil
2015-07-21 17:29:25 +00:00
}
func shouldGatherNode(node Node, serv *Server) bool {
if len(serv.Nodes) == 0 {
return true
}
for _, name := range serv.Nodes {
if name == node.Name {
return true
}
}
return false
}
func shouldGatherQueue(queue Queue, serv *Server) bool {
if len(serv.Queues) == 0 {
return true
2015-07-21 17:29:25 +00:00
}
for _, name := range serv.Queues {
if name == queue.Name {
return true
}
2015-07-21 17:29:25 +00:00
}
return false
2015-07-21 17:29:25 +00:00
}
func init() {
plugins.Add("rabbitmq", func() plugins.Plugin {
return &RabbitMQ{}
})
}