From fdcb6648c6326ad7d8e2c285d3ba35285a5c7c93 Mon Sep 17 00:00:00 2001
From: zhuojie <zhuojie@checkr.com>
Date: Tue, 9 Oct 2018 17:04:21 -0700
Subject: [PATCH] Add kafka data recorder dd metrics

---
 pkg/handler/data_recorder_kafka.go | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/pkg/handler/data_recorder_kafka.go b/pkg/handler/data_recorder_kafka.go
index c838f00f..7a9fb512 100644
--- a/pkg/handler/data_recorder_kafka.go
+++ b/pkg/handler/data_recorder_kafka.go
@@ -5,6 +5,7 @@ import (
 	"crypto/x509"
 	"encoding/base64"
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"strings"
 	"time"
@@ -106,6 +107,21 @@ func (k *kafkaRecorder) AsyncRecord(r *models.EvalResult) {
 		Value:     kr,
 		Timestamp: time.Now().UTC(),
 	}
+
+	logKafkaAsyncRecordToDatadog(r)
+}
+
+var logKafkaAsyncRecordToDatadog = func(r *models.EvalResult) {
+	if config.Global.StatsdClient == nil {
+		return
+	}
+	config.Global.StatsdClient.Incr(
+		"data_recorder.kafka",
+		[]string{
+			fmt.Sprintf("FlagID:%d", util.SafeUint(r.FlagID)),
+		},
+		float64(1),
+	)
 }
 
 type kafkaEvalResult struct {