From 61fe000eca8b9a119bbc721976496c97d408a14e Mon Sep 17 00:00:00 2001
From: Eric Karge <eric.karge@meteogroup.com>
Date: Wed, 8 Mar 2017 11:28:24 +0100
Subject: [PATCH] Fix topic label for Kafka metrics

---
 amqp-to-kafka.go        | 2 +-
 metrics.go              | 4 ++--
 test/docker-compose.yml | 1 -
 3 files changed, 3 insertions(+), 4 deletions(-)

diff --git a/amqp-to-kafka.go b/amqp-to-kafka.go
index 948d122..d7559c2 100644
--- a/amqp-to-kafka.go
+++ b/amqp-to-kafka.go
@@ -93,7 +93,7 @@ func forwardToKafka(delivery amqp.Delivery, kafkaProducer Producer) (forwarded [
     messageCounter.WithLabelValues("skipped").Inc()
   } else {
     forwarded = append(forwarded, delivery)
-    kafkaOffsets.WithLabelValues(strconv.FormatInt(int64(partition), 10)).Set(float64(offset))
+    kafkaOffsets.WithLabelValues(kafkaTopic, strconv.FormatInt(int64(partition), 10)).Set(float64(offset))
     messageCounter.WithLabelValues("forwarded").Inc()
   }
   return
diff --git a/metrics.go b/metrics.go
index 4cf6a51..ad6f1e4 100644
--- a/metrics.go
+++ b/metrics.go
@@ -24,8 +24,8 @@ import (
 var messageCounter = prometheus.NewCounterVec(prometheus.CounterOpts{Name:"messages", Help: "Messages handled"}, []string{"state"})
 var kafkaOffsets = prometheus.NewGaugeVec(prometheus.GaugeOpts{
   Name:"kafka_offset", Help: "Last known Kafka offsets",
-  ConstLabels: prometheus.Labels{"topic": kafkaTopic, "direction": "produced"},
-}, []string{"partition"})
+  ConstLabels: prometheus.Labels{"direction": "produced"},
+}, []string{"topic", "partition"})
 
 func startPrometheusHttpExporter() {
   if metricsAddress == "" {
diff --git a/test/docker-compose.yml b/test/docker-compose.yml
index 37e3fe9..986955b 100644
--- a/test/docker-compose.yml
+++ b/test/docker-compose.yml
@@ -11,7 +11,6 @@ services:
     environment:
     - AMQP_URI=amqp://amqp
     - AMQP_EXCHANGE=public
-    - AMQP_QUEUE=
     - AMQP_BINDING_KEY=/messages
     - AMQP_CONSUMER_TAG=my-tag
     - KAFKA_BROKERS=kafka:9092