Fix dropwizard parsing of metric names (elastic#8385)

Dropwizard metrics parser extracts tags embedded into metrics names.
However, it crashes when the metric name contains curly braces.

This patch improves parsing by adding additional checks and also makes
sures that only tags at the end of the metric name are extracted.

Assuming that the tags extracted by the dropwizard module are the ones
introduced by this PR: dropwizard/metrics#839

Fixes elastic#8365

CHANGELOG.asciidoc

@@ -78,6 +78,7 @@ https://github.com/elastic/beats/compare/v6.4.0...master[Check the HEAD diff]
 - Fixed the location of the modules.d dir in Deb and RPM packages. {issue}8104[8104]
 - Add docker diskio stats on Windows. {issue}6815[6815] {pull}8126[8126]
 - Fix incorrect type conversion of average response time in Haproxy dashboards {pull}8404[8404]
+- Fix dropwizard module parsing of metric names. {issue}8365[8365] {pull}6385[8385]
 
 *Packetbeat*
 

metricbeat/module/dropwizard/collector/data.go

@@ -92,8 +92,12 @@ func splitTagsFromMetricName(metricName string) (string, common.MapStr) {
 	if metricName == "" {
 		return "", nil
 	}
+	// Tags are located at the end
+	if metricName[len(metricName)-1] != '}' {
+		return metricName, nil
+	}
 
-	index := strings.Index(metricName, "{")
+	index := strings.LastIndex(metricName, "{")
 	if index == -1 {
 		return metricName, nil
 	}
@@ -104,20 +108,23 @@ func splitTagsFromMetricName(metricName string) (string, common.MapStr) {
 	tagStr := metricName[index+1 : len(metricName)-1]
 
 	for {
-		dobreak := false
 		ind := strings.Index(tagStr, ",")
 		eqPos := strings.Index(tagStr, "=")
+		if eqPos == -1 || ind != -1 && eqPos > ind {
+			return metricName, nil
+		}
 		if ind == -1 {
 			tags[tagStr[:eqPos]] = tagStr[eqPos+1:]
-			dobreak = true
-		} else {
-			tags[tagStr[:eqPos]] = tagStr[eqPos+1 : ind]
-			tagStr = tagStr[ind+2:]
+			break
 		}
-
-		if dobreak == true {
+		tags[tagStr[:eqPos]] = tagStr[eqPos+1 : ind]
+		if ind+2 >= len(tagStr) {
 			break
 		}
+		tagStr = tagStr[ind+1:]
+		if tagStr[0] == ' ' {
+			tagStr = tagStr[1:]
+		}
 	}
 
 	return key, tags

metricbeat/module/dropwizard/collector/data_test.go

@@ -0,0 +1,92 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package collector
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/libbeat/common"
+)
+
+func TestSplitTagsFromMetricName(t *testing.T) {
+	for _, testCase := range []struct {
+		title string
+		name  string
+		key   string
+		tags  common.MapStr
+	}{
+		{
+			title: "no tags",
+			name:  "my_metric1",
+		}, {
+			title: "parameter",
+			name:  "metric/{something}/other",
+		}, {
+			title: "trailing parameter",
+			name:  "metric/{notakey}",
+		}, {
+			title: "standard tags",
+			name:  "metric{key1=var1, key2=var2}",
+			key:   "metric",
+			tags:  common.MapStr{"key1": "var1", "key2": "var2"},
+		}, {
+			title: "standard tags (no space)",
+			name:  "metric{key1=var1,key2=var2}",
+			key:   "metric",
+			tags:  common.MapStr{"key1": "var1", "key2": "var2"},
+		}, {
+			title: "empty parameter",
+			name:  "metric/{}",
+		}, {
+			title: "empty key or value",
+			name:  "metric{=var1, key2=}",
+			key:   "metric",
+			tags:  common.MapStr{"": "var1", "key2": ""},
+		}, {
+			title: "empty key and value",
+			name:  "metric{=}",
+			key:   "metric",
+			tags:  common.MapStr{"": ""},
+		}, {
+			title: "extra comma",
+			name:  "metric{a=b,}",
+			key:   "metric",
+			tags:  common.MapStr{"a": "b"},
+		}, {
+			title: "extra comma and space",
+			name:  "metric{a=b, }",
+			key:   "metric",
+			tags:  common.MapStr{"a": "b"},
+		}, {
+			title: "extra comma and space",
+			name:  "metric{,a=b}",
+		},
+	} {
+		t.Run(testCase.title, func(t *testing.T) {
+			key, tags := splitTagsFromMetricName(testCase.name)
+			if testCase.key == "" && tags == nil {
+				assert.Equal(t, testCase.name, key)
+			} else {
+				assert.Equal(t, testCase.key, key)
+				assert.Equal(t, testCase.tags, tags)
+			}
+		})
+	}
+}