added self label to queues and connections

bertmap.go

@@ -12,19 +12,17 @@ import (
 // implementation) allow parsing of BERT-encoded RabbitMQ replies in a
 // way that's fully compatible with JSON parser from jsonmap.go
 type rabbitBERTReply struct {
-	body []byte
+	body    []byte
+	objects bert.Term
 }
 
-func makeBERTReply(body []byte) RabbitReply {
-	return &rabbitBERTReply{body}
+func makeBERTReply(body []byte) (RabbitReply, error) {
+	rawObjects, err := bert.Decode(body)
+	return &rabbitBERTReply{body, rawObjects}, err
 }
 
 func (rep *rabbitBERTReply) MakeStatsInfo(labels []string) []StatsInfo {
-	rawObjects, err := bert.Decode(rep.body)
-	if err != nil {
-		log.WithField("error", err).Error("Error while decoding bert")
-		return make([]StatsInfo, 0)
-	}
+	rawObjects := rep.objects
 
 	objects, ok := rawObjects.([]bert.Term)
 	if !ok {
@@ -48,12 +46,7 @@ func (rep *rabbitBERTReply) MakeStatsInfo(labels []string) []StatsInfo {
 
 func (rep *rabbitBERTReply) MakeMap() MetricMap {
 	flMap := make(MetricMap)
-	term, err := bert.Decode(rep.body)
-
-	if err != nil {
-		log.WithField("error", err).Error("Error while decoding bert")
-		return flMap
-	}
+	term := rep.objects
 
 	parseProplist(&flMap, "", term)
 	return flMap
@@ -321,3 +314,25 @@ func (err *bertDecodeError) Error() string {
 func bertError(message string, object interface{}) error {
 	return &bertDecodeError{message, object}
 }
+
+func (rep *rabbitBERTReply) GetString(label string) (string, bool) {
+	var resValue string
+	var result bool
+	result = false
+
+	iterateBertKV(rep.objects, func(key string, value interface{}) bool {
+		//Check if current key should be saved as label
+
+		if key == label {
+			tmp, ok := parseBertStringy(value)
+			if !ok {
+				return false
+			}
+			resValue = tmp
+			result = true
+			return false
+		}
+		return true
+	})
+	return resValue, result
+}

bertmap_test.go

@@ -53,8 +53,8 @@ func assertBertStatsEquivalence(t *testing.T, baseFileName string, labels []stri
 	t.Helper()
 	json, bert := tryReadFiles(t, baseFileName, "json", "bert")
 
-	jsonReply := makeJSONReply(json)
-	bertReply := makeBERTReply(bert)
+	jsonReply, _ := makeJSONReply(json)
+	bertReply, _ := makeBERTReply(bert)
 
 	bertParsed := bertReply.MakeStatsInfo(labels)
 	jsonParsed := jsonReply.MakeStatsInfo(labels)
@@ -67,8 +67,8 @@ func assertBertStatsEquivalence(t *testing.T, baseFileName string, labels []stri
 func assertBertMetricMapEquivalence(t *testing.T, baseFileName string) {
 	json, bert := tryReadFiles(t, baseFileName, "json", "bert")
 
-	jsonReply := makeJSONReply(json)
-	bertReply := makeBERTReply(bert)
+	jsonReply, _ := makeJSONReply(json)
+	bertReply, _ := makeBERTReply(bert)
 
 	bertParsed := bertReply.MakeMap()
 	jsonParsed := jsonReply.MakeMap()

decoder.go

@@ -25,13 +25,17 @@ type RabbitReply interface {
 	// RabbitMQ objects (i.e. list of queues, exchanges, etc.).
 	// Failure to parse should result in an empty result list.
 	MakeStatsInfo([]string) []StatsInfo
+
+	// GetString returns the string value for the given key
+	// If the key cannot be found the second return is false
+	GetString(key string) (string, bool)
 }
 
 // MakeReply instantiates the apropriate reply parser for a given
 // reply and the current configuration.
 func MakeReply(config rabbitExporterConfig, body []byte) (RabbitReply, error) {
 	if isCapEnabled(config, rabbitCapBert) {
-		return makeBERTReply(body), nil
+		return makeBERTReply(body)
 	}
-	return makeJSONReply(body), nil
+	return makeJSONReply(body)
 }

exporter.go

@@ -89,6 +89,7 @@ func (e *exporter) Collect(ch chan<- prometheus.Metric) {
 	ctx = context.WithValue(ctx, endpointScrapeDuration, e.endpointScrapeDurationMetric)
 	ctx = context.WithValue(ctx, endpointUpMetric, e.endpointUpMetric)
 	if err := collectWithDuration(ctx, e.overviewExporter, "overview", ch); err != nil {
+		log.WithError(err).Warn("retrieving overview failed")
 		allUp = false
 	}
 
@@ -97,6 +98,7 @@ func (e *exporter) Collect(ch chan<- prometheus.Metric) {
 	for name, ex := range e.exporter {
 
 		if err := collectWithDuration(ctx, ex, name, ch); err != nil {
+			log.WithError(err).Warn("retrieving " + name + " failed")
 			allUp = false
 		}
 	}

exporter_connections.go

@@ -11,9 +11,9 @@ func init() {
 }
 
 var (
-	connectionLabels            = []string{"vhost", "node", "peer_host", "user"}
-	connectionLabelsStateMetric = []string{"vhost", "node", "peer_host", "user", "state"}
-	connectionLabelKeys         = []string{"vhost", "node", "peer_host", "user", "state"}
+	connectionLabels            = []string{"vhost", "node", "peer_host", "user", "self"}
+	connectionLabelsStateMetric = []string{"vhost", "node", "peer_host", "user", "state", "self"}
+	connectionLabelKeys         = []string{"vhost", "node", "peer_host", "user", "state", "node"}
 
 	connectionGaugeVec = map[string]*prometheus.GaugeVec{
 		"channels":  newGaugeVec("connection_channels", "number of channels in use", connectionLabels),
@@ -52,17 +52,30 @@ func (e exporterConnections) Collect(ctx context.Context, ch chan<- prometheus.M
 	}
 	e.stateMetric.Reset()
 
+	selfNode := ""
+	if n, ok := ctx.Value(nodeName).(string); ok {
+		selfNode = n
+	}
+
 	for key, gauge := range e.metricsGV {
 		for _, connD := range connectionData {
 			if value, ok := connD.metrics[key]; ok {
-				gauge.WithLabelValues(connD.labels["vhost"], connD.labels["node"], connD.labels["peer_host"], connD.labels["user"]).Add(value)
+				self := "0"
+				if connD.labels["node"] == selfNode {
+					self = "1"
+				}
+				gauge.WithLabelValues(connD.labels["vhost"], connD.labels["node"], connD.labels["peer_host"], connD.labels["user"], self).Add(value)
 			}
 		}
 	}
 
 	for _, connD := range connectionData {
 		if _, ok := connD.metrics["channels"]; ok { // "channels" is used to retrieve one record per connection for setting the state
-			e.stateMetric.WithLabelValues(connD.labels["vhost"], connD.labels["node"], connD.labels["peer_host"], connD.labels["user"], connD.labels["state"]).Add(1)
+			self := "0"
+			if connD.labels["node"] == selfNode {
+				self = "1"
+			}
+			e.stateMetric.WithLabelValues(connD.labels["vhost"], connD.labels["node"], connD.labels["peer_host"], connD.labels["user"], connD.labels["state"], self).Add(1)
 		}
 	}
 

exporter_overview.go

@@ -1,10 +1,7 @@
 package main
 
 import (
-	"bytes"
 	"context"
-	"encoding/json"
-	"io"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/prometheus/client_golang/prometheus"
@@ -58,17 +55,15 @@ func (e *exporterOverview) Collect(ctx context.Context, ch chan<- prometheus.Met
 		return err
 	}
 
-	dec := json.NewDecoder(bytes.NewReader(body))
-	var nodeInfo NodeInfo
-	if err := dec.Decode(&nodeInfo); err == io.EOF {
-		return err
-	}
-	e.nodeInfo = nodeInfo
-
 	reply, err := MakeReply(config, body)
 	if err != nil {
 		return err
 	}
+
+	e.nodeInfo.Node, _ = reply.GetString("node")
+	e.nodeInfo.ErlangVersion, _ = reply.GetString("erlang_version")
+	e.nodeInfo.RabbitmqVersion, _ = reply.GetString("rabbitmq_version")
+
 	rabbitMqOverviewData := reply.MakeMap()
 
 	log.WithField("overviewData", rabbitMqOverviewData).Debug("Overview data")

exporter_queue.go

@@ -13,8 +13,8 @@ func init() {
 }
 
 var (
-	queueLabels    = []string{"vhost", "queue", "durable", "policy"}
-	queueLabelKeys = []string{"vhost", "name", "durable", "policy", "state"}
+	queueLabels    = []string{"vhost", "queue", "durable", "policy", "self"}
+	queueLabelKeys = []string{"vhost", "name", "durable", "policy", "state", "node"}
 
 	queueGaugeVec = map[string]*prometheus.GaugeVec{
 		"messages_ready":               newGaugeVec("queue_messages_ready", "Number of messages ready to be delivered to clients.", queueLabels),
@@ -93,6 +93,11 @@ func (e exporterQueue) Collect(ctx context.Context, ch chan<- prometheus.Metric)
 		}
 	}
 
+	selfNode := ""
+	if n, ok := ctx.Value(nodeName).(string); ok {
+		selfNode = n
+	}
+
 	rabbitMqQueueData, err := getStatsInfo(config, "queues", queueLabelKeys)
 
 	if err != nil {
@@ -110,8 +115,12 @@ func (e exporterQueue) Collect(ctx context.Context, ch chan<- prometheus.Metric)
 					if skipVhost := config.SkipVHost.MatchString(strings.ToLower(vname)); !skipVhost {
 						if matchInclude := config.IncludeQueues.MatchString(strings.ToLower(qname)); matchInclude {
 							if matchSkip := config.SkipQueues.MatchString(strings.ToLower(qname)); !matchSkip {
+								self := "0"
+								if queue.labels["node"] == selfNode {
+									self = "1"
+								}
 								// log.WithFields(log.Fields{"vhost": queue.labels["vhost"], "queue": queue.labels["name"], "key": key, "value": value}).Info("Set queue metric for key")
-								gaugevec.WithLabelValues(queue.labels["vhost"], queue.labels["name"], queue.labels["durable"], queue.labels["policy"]).Set(value)
+								gaugevec.WithLabelValues(queue.labels["vhost"], queue.labels["name"], queue.labels["durable"], queue.labels["policy"], self).Set(value)
 							}
 						}
 					}
@@ -129,7 +138,11 @@ func (e exporterQueue) Collect(ctx context.Context, ch chan<- prometheus.Metric)
 				if skipVhost := config.SkipVHost.MatchString(strings.ToLower(vname)); !skipVhost {
 					if matchInclude := config.IncludeQueues.MatchString(strings.ToLower(qname)); matchInclude {
 						if matchSkip := config.SkipQueues.MatchString(strings.ToLower(qname)); !matchSkip {
-							e.stateMetric.WithLabelValues(queue.labels["vhost"], queue.labels["name"], queue.labels["durable"], queue.labels["policy"], queue.labels["state"]).Set(1)
+							self := "0"
+							if queue.labels["node"] == selfNode {
+								self = "1"
+							}
+							e.stateMetric.WithLabelValues(queue.labels["vhost"], queue.labels["name"], queue.labels["durable"], queue.labels["policy"], self, queue.labels["state"]).Set(1)
 						}
 					}
 				}
@@ -146,10 +159,14 @@ func (e exporterQueue) Collect(ctx context.Context, ch chan<- prometheus.Metric)
 				if skipVhost := config.SkipVHost.MatchString(strings.ToLower(vname)); !skipVhost {
 					if matchInclude := config.IncludeQueues.MatchString(strings.ToLower(qname)); matchInclude {
 						if matchSkip := config.SkipQueues.MatchString(strings.ToLower(qname)); !matchSkip {
+							self := "0"
+							if queue.labels["node"] == selfNode {
+								self = "1"
+							}
 							if value, ok := queue.metrics[key]; ok {
-								ch <- prometheus.MustNewConstMetric(countvec, prometheus.CounterValue, value, queue.labels["vhost"], queue.labels["name"], queue.labels["durable"], queue.labels["policy"])
+								ch <- prometheus.MustNewConstMetric(countvec, prometheus.CounterValue, value, queue.labels["vhost"], queue.labels["name"], queue.labels["durable"], queue.labels["policy"], self)
 							} else {
-								ch <- prometheus.MustNewConstMetric(countvec, prometheus.CounterValue, 0, queue.labels["vhost"], queue.labels["name"], queue.labels["durable"], queue.labels["policy"])
+								ch <- prometheus.MustNewConstMetric(countvec, prometheus.CounterValue, 0, queue.labels["vhost"], queue.labels["name"], queue.labels["durable"], queue.labels["policy"], self)
 							}
 						}
 					}