Merge branch '2ndprom' into 'master'

zweiter prometheus endpunkt, metrik name fixes

See merge request bwnetflow/kafka/consumer_dashboard!2

kafka-listener.go

@@ -97,6 +97,6 @@ func handleControlMessages() {
 		offsetDiff := ctrlMsg.Offset - offsetPerPartition[partition]
 		offsetPerPartition[partition] = ctrlMsg.Offset
 
-		promExporter.IncrementCtrl(*kafkaInTopic, partition, offsetDiff)
+		promExporterMeta.IncrementCtrl(*kafkaInTopic, partition, offsetDiff)
 	}
 }

main.go

@@ -19,6 +19,7 @@ import (
 // KafkaConn holds the global kafka connection
 var kafkaConn = kafka.Connector{}
 var promExporter = prometheus.Exporter{}
+var promExporterMeta = prometheus.Exporter{}
 var tophostExporter = tophost.Exporter{}
 
 func main() {

prometheus/collectors.go

@@ -2,7 +2,25 @@ package prometheus
 
 import "github.com/prometheus/client_golang/prometheus"
 
-var (
+var ( // Meta Monitoring Data, to be added to default /metrics
+	msgcount = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "kafka_messages_total",
+			Help: "Number of Kafka messages",
+		})
+
+	kafkalabels = []string{
+		"topic",
+		"partition",
+	}
+	kafkaOffsets = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "kafka_offset_current",
+			Help: "Current Kafka Offset of the consumer",
+		}, kafkalabels)
+)
+
+var ( // Flow Data, to be exported on /flowdata
 	labels = []string{
 		// "src_port",
 		// "dst_port",
@@ -12,18 +30,12 @@ var (
 		"direction",
 		"cid",
 		"peer",
-		"remotecountry",
+		// "remotecountry",
 	}
 
-	msgcount = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
-			Name: "msgcount",
-			Help: "Number Kafka messages.",
-		}, labels)
-
 	flowNumber = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
-			Name: "flow_number",
+			Name: "flow_number_total",
 			Help: "Number of Flows received.",
 		}, labels)
 	flowBytes = prometheus.NewCounterVec(
@@ -36,10 +48,8 @@ var (
 			Name: "flow_packets",
 			Help: "Number of Packets received across Flows.",
 		}, labels)
-)
 
-// TOP HOSTS
-var (
+	// TOP HOSTS
 	hostlabels = []string{
 		"cid",
 		"ipSrc",
@@ -55,20 +65,7 @@ var (
 
 	hostConnections = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
-			Name: "host_connections",
+			Name: "host_connections_total",
 			Help: "Number of Connections for top N hosts.",
 		}, hostlabels)
 )
-
-// KAFKA METRICS
-var (
-	kafkalabels = []string{
-		"topic",
-		"partition",
-	}
-	kafkaOffsets = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
-			Name: "kafka_offset",
-			Help: "Kafka Offset of the consumer",
-		}, kafkalabels)
-)

prometheus/exporter.go

@@ -12,20 +12,18 @@ import (
 type Exporter struct {
 }
 
-// Initialize Prometheus Exporter, listen on addr with path /metrics
+// Initialize Prometheus Exporter, listen on addr with path /metrics and /flowdata
 func (exporter *Exporter) Initialize(addr string) {
-	// export prometheus metrics
+	prometheus.MustRegister(msgcount, kafkaOffsets)
+
+	flowReg := prometheus.NewRegistry()
+	flowReg.MustRegister(flowNumber, flowBytes, flowPackets, hostBytes, hostConnections)
+
 	http.Handle("/metrics", promhttp.Handler())
+	http.Handle("/flowdata", promhttp.HandlerFor(flowReg, promhttp.HandlerOpts{}))
+
 	go func() {
 		http.ListenAndServe(addr, nil)
 	}()
-	log.Println("Enabled Prometheus metrics endpoint.")
-
-	// register collectors
-	exporter.registerCollectors()
-}
-
-func (exporter *Exporter) registerCollectors() {
-	// TODO make more dynamic
-	prometheus.MustRegister(msgcount, flowNumber, flowBytes, flowPackets, hostBytes, hostConnections, kafkaOffsets)
+	log.Println("Enabled Prometheus /metrics and /flowdata endpoints.")
 }

prometheus/increment.go

@@ -22,16 +22,16 @@ func (exporter *Exporter) Increment(flow *flow.FlowMessage) {
 	labels := prometheus.Labels{
 		// "src_port":      fmt.Sprint(srcPort),
 		// "dst_port":      fmt.Sprint(dstPort),
-		"ipversion":     flow.GetIPversion().String(),
-		"application":   application,
-		"protoname":     fmt.Sprint(flow.GetProtoName()),
-		"direction":     fmt.Sprint(flow.GetDirection()),
-		"cid":           fmt.Sprint(flow.GetCid()),
-		"peer":          flow.GetPeer(),
-		"remotecountry": flow.GetRemoteCountry(),
+		"ipversion":   flow.GetIPversion().String(),
+		"application": application,
+		"protoname":   fmt.Sprint(flow.GetProtoName()),
+		"direction":   fmt.Sprint(flow.GetDirection()),
+		"cid":         fmt.Sprint(flow.GetCid()),
+		"peer":        flow.GetPeer(),
+		// "remotecountry": flow.GetRemoteCountry(),
 	}
 
-	msgcount.With(labels).Inc()
+	msgcount.Inc()
 	flowNumber.With(labels).Add(float64(flow.GetSamplingRate()))
 	flowPackets.With(labels).Add(float64(flow.GetPackets()))
 	flowBytes.With(labels).Add(float64(flow.GetBytes()))