statsdbeat.diff

diff --git a/beater/statsdparser.go b/beater/statsdparser.go
index 24727e7..6bf5c65 100644
--- a/beater/statsdparser.go
+++ b/beater/statsdparser.go
@@ -33,8 +33,8 @@ func ParseBeats(msg string) ([]beat.Event, error) {
 
 func parseBeat(msg string) ([]beat.Event, error) {
 	parts := strings.Split(msg, "|")
-	if len(parts) < 2 || len(parts) > 3 {
-		return nil, fmt.Errorf("Expecting 2 or 3 parts of | but was %d", len(parts))
+	if len(parts) < 2 || len(parts) > 4 {
+		return nil, fmt.Errorf("Expecting 2-3 parts of | but was %d", len(parts))
 	}
 
 	//parts[0] has structure of  <bucket>(,<k>=<v>)*:<value>
@@ -43,6 +43,9 @@ func parseBeat(msg string) ([]beat.Event, error) {
 		return nil, err
 	}
 
+  //has the structure of #(,<k>:<v>|,<k>)
+  datadogTags := getDatadogTagsValues(parts[len(parts)-1])
+
 	_type := strings.TrimSpace(parts[1])
 	e := &beat.Event{
 		Timestamp: time.Now(),
@@ -65,6 +68,9 @@ func parseBeat(msg string) ([]beat.Event, error) {
 	if len(tags) > 0 {
 		bucketMap.Put("statsd.ctx", tags)
 	}
+  if len(datadogTags) > 0 {
+		bucketMap.Put("statsd.ctx", datadogTags)
+  }
 
 	switch _type {
 	case "c":
@@ -133,6 +139,25 @@ func getBucketTagsValue(part string) (bucket string, tags map[string]interface{}
 	return bucket, tags, val, err
 }
 
+func getDatadogTagsValues(part string) (tags map[string]interface{}) {
+  if ! strings.HasPrefix(part, "#") || len(part) < 2 {
+    return make(map[string]interface{})
+  }
+
+  parts := strings.Split(part[1:], ",")
+  tags = make(map[string]interface{}, len(parts))
+  for i := 0; i < len(parts); i++ {
+    kv := strings.Split(parts[i], ":")
+    if len(kv) == 2 {
+      tags[kv[0]] = kv[1]
+    } else {
+      tags[kv[0]] = kv[0]
+    }
+  }
+
+  return tags
+}
+
 //accounts.authentication.password.failure.no_email_found
 // We always have a target, then action, then section then namespace
 func splitBucket(bucket string) (namespace string, section string, target string, action string) {
diff --git a/beater/statsdparser_test.go b/beater/statsdparser_test.go
index 3483002..3bfc6f3 100644
--- a/beater/statsdparser_test.go
+++ b/beater/statsdparser_test.go
@@ -142,6 +142,25 @@ func Test_parseBeat(t *testing.T) {
 			},
 			false,
 		},
+		{"testWithDatadogStyleTags",
+			args{"myCounter:1|c|#myTag:error,tagB:2,tagC"},
+			[]beat.Event{
+				beat.Event{
+					Fields: common.MapStr{
+						"statsd.bucket": "myCounter",
+						"statsd.target": "myCounter",
+						"statsd.type":   "counter",
+						"statsd.value":  1,
+						"statsd.ctx": map[string]interface{}{
+							"myTag": "error",
+							"tagB":  "2",
+ 							"tagC": "tagC"
+						},
+					},
+				},
+			},
+			false,
+		},
 		{"testGauge",
 			args{"platform-insights.test.gauge.num_goroutine:4|g"},
 			[]beat.Event{