Merge pull request #207 from stripe/yasha-fix-set

Fix bugs with SSF parsing & add tests

CHANGELOG.md

@@ -1,3 +1,12 @@
+# 1.5.2, pending
+
+## Bugfixes
+* Correctly parse `Set` metrics if sent via SSF. Thanks [redsn0w422](https://github.com/redsn0w422)!
+* Return correct array of tags after parsing an SSF metric. Thanks [redsn0w422](https://github.com/redsn0w422)!
+
+## Improvements
+* Added tests for `parseMetricSSF`. Thanks [redsn0w422](https://github.com/redsn0w422)!
+
 # 1.5.1, 2017-07-18
 
 ## Bugfixes

parser_test.go

@@ -5,8 +5,86 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stripe/veneur/samplers"
+	"github.com/stripe/veneur/ssf"
 )
 
+func freshSSFMetric() *ssf.SSFSample {
+	return &ssf.SSFSample{
+		Metric:     0,
+		Name:       "test.ssf_metric",
+		Value:      5,
+		Message:    "test_msg",
+		Status:     0,
+		SampleRate: 1,
+		Tags: map[string]string{
+			"tag1": "value1",
+			"tag2": "value2",
+		},
+	}
+}
+
+func TestParserSSF(t *testing.T) {
+	standardMetric := freshSSFMetric()
+	m, _ := samplers.ParseMetricSSF(standardMetric)
+	assert.NotNil(t, m, "Got nil metric!")
+	assert.Equal(t, standardMetric.Name, m.Name, "Name")
+	assert.Equal(t, float64(standardMetric.Value), m.Value, "Value")
+	assert.Equal(t, "counter", m.Type, "Type")
+}
+
+func TestParserSSFGauge(t *testing.T) {
+	standardMetric := freshSSFMetric()
+	standardMetric.Metric = ssf.SSFSample_GAUGE
+	m, _ := samplers.ParseMetricSSF(standardMetric)
+	assert.NotNil(t, m, "Got nil metric!")
+	assert.Equal(t, standardMetric.Name, m.Name, "Name")
+	assert.Equal(t, float64(standardMetric.Value), m.Value, "Value")
+	assert.Equal(t, "gauge", m.Type, "Type")
+}
+
+func TestParserSSFSet(t *testing.T) {
+	standardMetric := freshSSFMetric()
+	standardMetric.Metric = ssf.SSFSample_SET
+	m, _ := samplers.ParseMetricSSF(standardMetric)
+	assert.NotNil(t, m, "Got nil metric!")
+	assert.Equal(t, standardMetric.Name, m.Name, "Name")
+	assert.Equal(t, standardMetric.Message, m.Value, "Value")
+	assert.Equal(t, "set", m.Type, "Type")
+}
+
+func TestParserSSFWithTags(t *testing.T) {
+	standardMetric := freshSSFMetric()
+	m, _ := samplers.ParseMetricSSF(standardMetric)
+	assert.NotNil(t, m, "Got nil metric!")
+	assert.Equal(t, standardMetric.Name, m.Name, "Name")
+	assert.Equal(t, float64(standardMetric.Value), m.Value, "Value")
+	assert.Equal(t, "counter", m.Type, "Type")
+	assert.Equal(t, 2, len(m.Tags), "# of Tags")
+}
+
+func TestParserSSFWithSampleRate(t *testing.T) {
+	standardMetric := freshSSFMetric()
+	standardMetric.SampleRate = 0.1
+	m, _ := samplers.ParseMetricSSF(standardMetric)
+	assert.NotNil(t, m, "Got nil metric!")
+	assert.Equal(t, standardMetric.Name, m.Name, "Name")
+	assert.Equal(t, float64(standardMetric.Value), m.Value, "Value")
+	assert.Equal(t, "counter", m.Type, "Type")
+	assert.Equal(t, float32(0.1), m.SampleRate, "Sample Rate")
+}
+
+func TestParserSSFWithSampleRateAndTags(t *testing.T) {
+	standardMetric := freshSSFMetric()
+	standardMetric.SampleRate = 0.1
+	m, _ := samplers.ParseMetricSSF(standardMetric)
+	assert.NotNil(t, m, "Got nil metric!")
+	assert.Equal(t, standardMetric.Name, m.Name, "Name")
+	assert.Equal(t, float64(standardMetric.Value), m.Value, "Value")
+	assert.Equal(t, "counter", m.Type, "Type")
+	assert.Equal(t, float32(0.1), m.SampleRate, "Sample Rate")
+	assert.Len(t, m.Tags, 2, "Tags")
+}
+
 func TestParser(t *testing.T) {
 	m, _ := samplers.ParseMetric([]byte("a.b.c:1|c"))
 	assert.NotNil(t, m, "Got nil metric!")

samplers/parser.go

@@ -72,9 +72,13 @@ func ParseMetricSSF(metric *ssf.SSFSample) (*UDPMetric, error) {
 		return nil, invalidMetricTypeError
 	}
 	h.Write([]byte(ret.Type))
-	ret.Value = float64(metric.Value)
+	if metric.Metric == ssf.SSFSample_SET {
+		ret.Value = metric.Message
+	} else {
+		ret.Value = float64(metric.Value)
+	}
 	ret.SampleRate = metric.SampleRate
-	tempTags := make([]string, len(metric.Tags))
+	tempTags := make([]string, 0)
 	for key, value := range metric.Tags {
 		if key == "veneurlocalonly" {
 			// delete the tag from the list

samplers/samplers_test.go

@@ -11,7 +11,6 @@ import (
 )
 
 func TestCounterEmpty(t *testing.T) {
-
 	c := NewCounter("a.b.c", []string{"a:b"})
 	c.Sample(1, 1.0)
 
@@ -32,7 +31,6 @@ func TestCounterEmpty(t *testing.T) {
 }
 
 func TestCounterRate(t *testing.T) {
-
 	c := NewCounter("a.b.c", []string{"a:b"})
 
 	c.Sample(5, 1.0)
@@ -43,7 +41,6 @@ func TestCounterRate(t *testing.T) {
 }
 
 func TestCounterSampleRate(t *testing.T) {
-
 	c := NewCounter("a.b.c", []string{"a:b"})
 
 	c.Sample(5, 0.5)
@@ -80,7 +77,6 @@ func TestCounterMerge(t *testing.T) {
 }
 
 func TestGauge(t *testing.T) {
-
 	g := NewGauge("a.b.c", []string{"a:b"})
 
 	assert.Equal(t, "a.b.c", g.Name, "Name")
@@ -155,7 +151,6 @@ func TestSetMerge(t *testing.T) {
 }
 
 func TestHisto(t *testing.T) {
-
 	h := NewHist("a.b.c", []string{"a:b"})
 
 	assert.Equal(t, "a.b.c", h.Name, "Name")
@@ -253,7 +248,6 @@ func TestHisto(t *testing.T) {
 }
 
 func TestHistoAvgOnly(t *testing.T) {
-
 	h := NewHist("a.b.c", []string{"a:b"})
 
 	assert.Equal(t, "a.b.c", h.Name, "Name")
@@ -290,7 +284,6 @@ func TestHistoAvgOnly(t *testing.T) {
 }
 
 func TestHistoSampleRate(t *testing.T) {
-
 	h := NewHist("a.b.c", []string{"a:b"})
 
 	assert.Equal(t, "a.b.c", h.Name, "Name")

server.go

@@ -547,6 +547,17 @@ func ValidTrace(sample ssf.SSFSpan) bool {
 	return ret
 }
 
+// ValidMetric takes in an SSF sample and determines if it is valid or not.
+func ValidMetric(sample samplers.UDPMetric) bool {
+	ret := true
+	ret = ret && sample.Name != ""
+	ret = ret && sample.Value != nil
+	if sample.SampleRate == 0 {
+		sample.SampleRate = 1
+	}
+	return ret
+}
+
 // HandleTracePacket accepts an incoming packet as bytes and sends it to the
 // appropriate worker.
 func (s *Server) HandleTracePacket(packet []byte) {
@@ -577,11 +588,17 @@ func (s *Server) HandleTracePacket(packet []byte) {
 			s.Statsd.Count("packet.error_total", 1, []string{"packet_type:ssf_metric", "reason:parse"}, 1.0)
 			return
 		}
-		s.Workers[metric.Digest%uint32(len(s.Workers))].PacketChan <- *metric
+		if ValidMetric(*metric) {
+			s.Workers[metric.Digest%uint32(len(s.Workers))].PacketChan <- *metric
+		} else {
+			s.Statsd.Count("packet.error_total", 1, []string{"packet_type:ssf_metric", "reason:not_valid"}, 1.0)
+		}
 	}
 	if ValidTrace(*newSample) {
 		s.Statsd.Incr("packet.spans.received_total", nil, .1)
 		s.TraceWorker.TraceChan <- *newSample
+	} else {
+		s.Statsd.Count("packet.error_total", 1, []string{"packet_type:ssf_span", "reason:not_valid"}, 1.0)
 	}
 }