BUGFIX/FEATURE: Sanitize StatsD jobs/events beacuse some StatsD servers can't accept keys with colons or pipes. Provide the option to write your own sanitizer in case your StatsD implementation doesn't suck.

Author: cypriss

statsd_sink.go

@@ -7,8 +7,12 @@ import (
 	"time"
 )
 
+type StatsDSinkSantizationFunc func(string) string
+
 // This sink emits to a StatsD deaemon by sending it a UDP packet.
 type StatsDSink struct {
+	SanitizationFunc StatsDSinkSantizationFunc
+
 	conn net.Conn
 
 	// Prefix is something like "metroid"
@@ -25,8 +29,9 @@ func NewStatsDSink(addr, prefix string) (Sink, error) {
 	}
 
 	sink := &StatsDSink{
-		conn:   c,
-		prefix: prefix,
+		SanitizationFunc: sanitizeKey,
+		conn:             c,
+		prefix:           prefix,
 	}
 
 	return sink, nil
@@ -65,7 +70,7 @@ func (s *StatsDSink) EmitComplete(job string, status CompletionStatus, nanos int
 		b.WriteString(s.prefix)
 		b.WriteRune('.')
 	}
-	b.WriteString(job)
+	b.WriteString(s.SanitizationFunc(job))
 	b.WriteRune('.')
 	b.WriteString(completionStatusToString[status])
 
@@ -83,10 +88,10 @@ func (s *StatsDSink) eventKeys(job, event, suffix string) (string, string) {
 		key2.WriteRune('.')
 	}
 
-	key1.WriteString(event)
-	key2.WriteString(job)
+	key1.WriteString(s.SanitizationFunc(event))
+	key2.WriteString(s.SanitizationFunc(job))
 	key2.WriteRune('.')
-	key2.WriteString(event)
+	key2.WriteString(s.SanitizationFunc(event))
 
 	if suffix != "" {
 		key1.WriteRune('.')
@@ -117,3 +122,15 @@ func (s *StatsDSink) measure(key string, nanos int64) {
 func (s *StatsDSink) send(msg []byte) {
 	s.conn.Write(msg)
 }
+
+func sanitizeKey(k string) string {
+	var key bytes.Buffer
+	for _, c := range k {
+		if c == '|' || c == ':' {
+			key.WriteRune('$')
+		} else {
+			key.WriteRune(c)
+		}
+	}
+	return key.String()
+}

statsd_sink_test.go

@@ -54,6 +54,14 @@ func TestStatsDSinkEmitEventPrefix(t *testing.T) {
 	})
 }
 
+func TestStatsDSinkEmitEventShouldSanitize(t *testing.T) {
+	sink, err := NewStatsDSink(testAddr, "metroid")
+	assert.NoError(t, err)
+	listenFor(t, []string{"metroid.my$event:1|c\n", "metroid.my$job.my$event:1|c\n"}, func() {
+		sink.EmitEvent("my|job", "my:event", nil)
+	})
+}
+
 func TestStatsDSinkEmitEventNoPrefix(t *testing.T) {
 	sink, err := NewStatsDSink(testAddr, "")
 	assert.NoError(t, err)