x-pack/metricbeat/module/statsd: Skip invalid statsd packet(s)

CHANGELOG.next.asciidoc

@@ -191,6 +191,7 @@ automatic splitting at root level, if root level element is an array. {pull}3415
 - Fix no error logs displayed in CloudWatch EC2, RDS and SQS metadata {issue}34985[34985] {pull}35035[35035]
 - Remove Beta warning from IIS application_pool metricset {pull}35480[35480]
 - Improve documentation for ActiveMQ module {issue}35113[35113] {pull}35558[35558]
+- Resolve statsd module's prematurely halting of metrics parsing upon encountering an invalid packet. {pull}35075[35075]
 
 *Osquerybeat*
 

x-pack/metricbeat/module/statsd/server/data.go

@@ -6,11 +6,11 @@ package server
 
 import (
 	"bytes"
+	"errors"
+	"fmt"
 	"strconv"
 	"time"
 
-	"github.com/pkg/errors"
-
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/metricbeat/helper/server"
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -20,8 +20,7 @@ import (
 var errInvalidPacket = errors.New("invalid statsd packet")
 
 type metricProcessor struct {
-	registry      *registry
-	reservoirSize int
+	registry *registry
 }
 
 type statsdMetric struct {
@@ -32,12 +31,12 @@ type statsdMetric struct {
 	tags       map[string]string
 }
 
-func splitTags(rawTags []byte, kvSep []byte) map[string]string {
+func splitTags(rawTags, kvSep []byte) map[string]string {
 	tags := map[string]string{}
 	for _, kv := range bytes.Split(rawTags, []byte(",")) {
 		kvSplit := bytes.SplitN(kv, kvSep, 2)
 		if len(kvSplit) != 2 {
-			logger.Warnf("could not parse tags")
+			logger.Warn("could not parse tags")
 			continue
 		}
 		tags[string(kvSplit[0])] = string(kvSplit[1])
@@ -86,14 +85,16 @@ func parseSingle(b []byte) (statsdMetric, error) {
 	return s, nil
 }
 
-// parse will parse a statsd metric into its components
+// parse will parse statsd metrics into individual metric and then its components
 func parse(b []byte) ([]statsdMetric, error) {
-	metrics := []statsdMetric{}
-	for _, rawMetric := range bytes.Split(b, []byte("\n")) {
-		if len(rawMetric) > 0 {
-			metric, err := parseSingle(rawMetric)
+	rawMetrics := bytes.Split(b, []byte("\n"))
+	metrics := make([]statsdMetric, 0, len(rawMetrics))
+	for i := range rawMetrics {
+		if len(rawMetrics[i]) > 0 {
+			metric, err := parseSingle(rawMetrics[i])
 			if err != nil {
-				return metrics, err
+				logger.Warnf("invalid packet: %s", err)
+				continue
 			}
 			metrics = append(metrics, metric)
 		}
@@ -120,13 +121,13 @@ func eventMapping(metricName string, metricValue interface{}, metricSetFields ma
 		// Not all labels match
 		// Skip and continue to next mapping
 		if len(res) != (len(mapping.Labels) + 1) {
-			logger.Debugf("not all labels match in statsd.mapping, skipped")
+			logger.Debug("not all labels match in statsd.mapping, skipped")
 			continue
 		}
 
 		// Let's add the metric set fields from labels
 		names := mapping.regex.SubexpNames()
-		for i, _ := range res {
+		for i := range res {
 			for _, label := range mapping.Labels {
 				if label.Attr != names[i] {
 					continue
@@ -139,8 +140,6 @@ func eventMapping(metricName string, metricValue interface{}, metricSetFields ma
 		// Let's add the metric with the value field
 		metricSetFields[mapping.Value.Field] = metricValue
 	}
-
-	return
 }
 
 func newMetricProcessor(ttl time.Duration) *metricProcessor {
@@ -162,10 +161,10 @@ func (p *metricProcessor) processSingle(m statsdMetric) error {
 		var err error
 		sampleRate, err = strconv.ParseFloat(m.sampleRate, 64)
 		if err != nil {
-			return errors.Wrapf(err, "failed to process metric `%s` sample rate `%s`", m.name, m.sampleRate)
+			return fmt.Errorf("failed to process metric `%s` sample rate `%s`: %w", m.name, m.sampleRate, err)
 		}
 		if sampleRate <= 0.0 {
-			return errors.Errorf("sample rate of 0.0 is invalid for metric `%s`", m.name)
+			return fmt.Errorf("sample rate of 0.0 is invalid for metric `%s`: %w", m.name, err)
 		}
 	}
 
@@ -174,7 +173,7 @@ func (p *metricProcessor) processSingle(m statsdMetric) error {
 		c := p.registry.GetOrNewCounter(m.name, m.tags)
 		v, err := strconv.ParseInt(m.value, 10, 64)
 		if err != nil {
-			return errors.Wrapf(err, "failed to process counter `%s` with value `%s`", m.name, m.value)
+			return fmt.Errorf("failed to process counter `%s` with value `%s`: %w", m.name, m.value, err)
 		}
 		// apply sample rate
 		v = int64(float64(v) * (1.0 / sampleRate))
@@ -183,9 +182,8 @@ func (p *metricProcessor) processSingle(m statsdMetric) error {
 		c := p.registry.GetOrNewGauge64(m.name, m.tags)
 		v, err := strconv.ParseFloat(m.value, 64)
 		if err != nil {
-			return errors.Wrapf(err, "failed to process gauge `%s` with value `%s`", m.name, m.value)
+			return fmt.Errorf("failed to process gauge `%s` with value `%s`: %w", m.name, m.value, err)
 		}
-
 		// inc/dec or set
 		if m.value[0] == '+' || m.value[0] == '-' {
 			c.Inc(v)
@@ -196,14 +194,14 @@ func (p *metricProcessor) processSingle(m statsdMetric) error {
 		c := p.registry.GetOrNewTimer(m.name, m.tags)
 		v, err := strconv.ParseFloat(m.value, 64)
 		if err != nil {
-			return errors.Wrapf(err, "failed to process timer `%s` with value `%s`", m.name, m.value)
+			return fmt.Errorf("failed to process timer `%s` with value `%s`: %w", m.name, m.value, err)
 		}
 		c.SampledUpdate(time.Duration(v), sampleRate)
 	case "h": // TODO: can these be floats?
 		c := p.registry.GetOrNewHistogram(m.name, m.tags)
 		v, err := strconv.ParseInt(m.value, 10, 64)
 		if err != nil {
-			return errors.Wrapf(err, "failed to process histogram `%s` with value `%s`", m.name, m.value)
+			return fmt.Errorf("failed to process histogram `%s` with value `%s`: %w", m.name, m.value, err)
 		}
 		c.Update(v)
 	case "s":

x-pack/metricbeat/module/statsd/server/data_test.go

@@ -5,7 +5,7 @@
 package server
 
 import (
-	"fmt"
+	"errors"
 	"testing"
 	"time"
 
@@ -820,7 +820,7 @@ func TestBuildMappings(t *testing.T) {
         value:
           field: started
 `,
-			err:      fmt.Errorf(`repeated label fields "repeated_label_field"`),
+			err:      errors.New(`repeated label fields "repeated_label_field"`),
 			expected: nil,
 		},
 		{
@@ -833,13 +833,15 @@ func TestBuildMappings(t *testing.T) {
         value:
           field: colliding_field
 `,
-			err:      fmt.Errorf(`collision between label field "colliding_field" and value field "colliding_field"`),
+			err:      errors.New(`collision between label field "colliding_field" and value field "colliding_field"`),
 			expected: nil,
 		},
 	} {
 		t.Run(test.title, func(t *testing.T) {
 			var mappings []StatsdMapping
-			err := yaml.Unmarshal([]byte(test.input), &mappings)
+			if err := yaml.Unmarshal([]byte(test.input), &mappings); err != nil {
+				t.Error(err)
+			}
 			actual, err := buildMappings(mappings)
 			for k, v := range actual {
 				v.regex = nil
@@ -883,12 +885,36 @@ func TestParseMetrics(t *testing.T) {
 			}},
 		},
 		{
-			input: "decrement-counter:-15|c",
-			expected: []statsdMetric{{
-				name:       "decrement-counter",
-				metricType: "c",
-				value:      "-15",
-			}},
+			// All metrics are parsed except the invalid packet
+			input: "decrement-counter:-15|c\nmeter1-1.4|m\ndecrement-counter:-20|c",
+			expected: []statsdMetric{
+				{
+					name:       "decrement-counter",
+					metricType: "c",
+					value:      "-15",
+				},
+				{
+					name:       "decrement-counter",
+					metricType: "c",
+					value:      "-20",
+				},
+			},
+		},
+		{
+			// All metrics are parsed except the invalid packet
+			input: "meter1-1.4|m\ndecrement-counter:-20|c\ntimer1:1.2|ms",
+			expected: []statsdMetric{
+				{
+					name:       "decrement-counter",
+					metricType: "c",
+					value:      "-20",
+				},
+				{
+					name:       "timer1",
+					metricType: "ms",
+					value:      "1.2",
+				},
+			},
 		},
 		{
 			input: "timer1:1.2|ms",
@@ -995,12 +1021,10 @@ func TestParseMetrics(t *testing.T) {
 		{
 			input:    "meter1-1.4|m",
 			expected: []statsdMetric{},
-			err:      errInvalidPacket,
 		},
 		{
 			input:    "meter1:1.4-m",
 			expected: []statsdMetric{},
-			err:      errInvalidPacket,
 		},
 	} {
 		actual, err := parse([]byte(test.input))
@@ -1016,6 +1040,47 @@ func TestParseMetrics(t *testing.T) {
 	}
 }
 
+func TestParseSingle(t *testing.T) {
+	tests := map[string]struct {
+		input string
+		err   error
+		want  statsdMetric
+	}{
+		"invalid packet #1": {input: "meter1-1.4|m", err: errInvalidPacket, want: statsdMetric{}},
+		"invalid packet #2": {input: "meter1:1.4-m", err: errInvalidPacket, want: statsdMetric{}},
+		"valid packet: counter with tags": {
+			input: "tags1:1|c|#k1:v1,k2:v2",
+			err:   nil,
+			want: statsdMetric{
+				name:       "tags1",
+				metricType: "c",
+				sampleRate: "",
+				value:      "1",
+				tags:       map[string]string{"k1": "v1", "k2": "v2"},
+			},
+		},
+		"valid packet: gauge": {
+			input: "gauge1:1.0|g",
+			err:   nil,
+			want: statsdMetric{
+				name:       "gauge1",
+				metricType: "g",
+				sampleRate: "",
+				value:      "1.0",
+				tags:       nil,
+			},
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			got, err := parseSingle([]byte(tc.input))
+			assert.Equal(t, tc.err, err)
+			assert.Equal(t, tc.want, got)
+		})
+	}
+}
+
 type testUDPEvent struct {
 	event mapstr.M
 	meta  server.Meta
@@ -1068,13 +1133,13 @@ func TestTagsGrouping(t *testing.T) {
 	}
 
 	expectedTags := []mapstr.M{
-		mapstr.M{
+		{
 			"labels": mapstr.M{
 				"k1": "v1",
 				"k2": "v2",
 			},
 		},
-		mapstr.M{
+		{
 			"labels": mapstr.M{
 				"k1": "v2",
 				"k2": "v3",
@@ -1182,6 +1247,7 @@ func TestGaugeDeltas(t *testing.T) {
 		"metric01": map[string]interface{}{"value": -1.0},
 	})
 }
+
 func TestCounter(t *testing.T) {
 	ms := mbtest.NewMetricSet(t, map[string]interface{}{"module": "statsd"}).(*MetricSet)
 	testData := []string{
@@ -1316,5 +1382,4 @@ func BenchmarkIngest(b *testing.B) {
 		err := ms.processor.Process(events[i%len(events)])
 		assert.NoError(b, err)
 	}
-
 }