add new stats (#79)

* add new stats

Signed-off-by: John Seekins <jseekins@datto.com>

* add debug logging about bad metrics

Signed-off-by: John Seekins <jseekins@datto.com>

Author: johnseekins

handlers.go

@@ -15,6 +15,7 @@
 package main
 
 import (
+	"fmt"
 	"io/ioutil"
 	"net/http"
 
@@ -68,14 +69,17 @@ func receiveHandler(producer *kafka.Producer, serializer Serializer) func(c *gin
 				Topic:     &t,
 			}
 			for _, metric := range metrics {
+				objectsWritten.Add(float64(1))
 				err := producer.Produce(&kafka.Message{
 					TopicPartition: part,
 					Value:          metric,
 				}, nil)
 
 				if err != nil {
+					objectsFailed.Add(float64(1))
 					c.AbortWithStatus(http.StatusInternalServerError)
-					logrus.WithError(err).Error("couldn't produce message in kafka")
+					logrus.WithError(err).Debug(fmt.Sprintf("Failing metric %v", metric))
+					logrus.WithError(err).Error(fmt.Sprintf("couldn't produce message in kafka topic %v", topic))
 					return
 				}
 			}

metrics.go

@@ -22,8 +22,44 @@ var (
 			Name: "http_requests_total",
 			Help: "Count of all http requests",
 		})
+	promBatches = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "incoming_prometheus_batches_total",
+			Help: "Count of incoming prometheus batches (to be broken into individual metrics)",
+		})
+	serializeTotal = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "serialized_total",
+			Help: "Count of all serialization requests",
+		})
+	serializeFailed = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "serialized_failed_total",
+			Help: "Count of all serialization failures",
+		})
+	objectsFiltered = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "objects_filtered_total",
+			Help: "Count of all filter attempts",
+		})
+	objectsWritten = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "objects_written_total",
+			Help: "Count of all objects written to Kafka",
+		})
+	objectsFailed = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "objects_failed_total",
+			Help: "Count of all objects write failures to Kafka",
+		})
 )
 
 func init() {
 	prometheus.MustRegister(httpRequestsTotal)
+	prometheus.MustRegister(promBatches)
+	prometheus.MustRegister(serializeTotal)
+	prometheus.MustRegister(serializeFailed)
+	prometheus.MustRegister(objectsFiltered)
+	prometheus.MustRegister(objectsFailed)
+	prometheus.MustRegister(objectsWritten)
 }

serializers.go

@@ -35,6 +35,7 @@ type Serializer interface {
 
 // Serialize generates the JSON representation for a given Prometheus metric.
 func Serialize(s Serializer, req *prompb.WriteRequest) (map[string][][]byte, error) {
+	promBatches.Add(float64(1))
 	result := make(map[string][][]byte)
 
 	for _, ts := range req.Timeseries {
@@ -49,6 +50,7 @@ func Serialize(s Serializer, req *prompb.WriteRequest) (map[string][][]byte, err
 		for _, sample := range ts.Samples {
 			name := string(labels["__name__"])
 			if !filter(name, labels) {
+				objectsFiltered.Add(float64(1))
 				continue
 			}
 
@@ -62,9 +64,10 @@ func Serialize(s Serializer, req *prompb.WriteRequest) (map[string][][]byte, err
 
 			data, err := s.Marshal(m)
 			if err != nil {
+				serializeFailed.Add(float64(1))
 				logrus.WithError(err).Errorln("couldn't marshal timeseries")
 			}
-
+			serializeTotal.Add(float64(1))
 			result[t] = append(result[t], data)
 		}
 	}