feat(inputs.logstash): Record number of failures

plugins/inputs/logstash/README.md

@@ -129,6 +129,7 @@ expose all stats)
     - duration_in_millis
     - in
     - out
+    - failures(if exists)
     - bulk_requests_failures (for Logstash 7+)
     - bulk_requests_with_errors (for Logstash 7+)
     - documents_successes (for logstash 7+)

plugins/inputs/logstash/logstash.go

@@ -92,6 +92,7 @@ type Plugin struct {
 	ID           string                 `json:"id"`
 	Events       interface{}            `json:"events"`
 	Name         string                 `json:"name"`
+	Failures     *int64                 `json:"failures,omitempty"`
 	BulkRequests map[string]interface{} `json:"bulk_requests"`
 	Documents    map[string]interface{} `json:"documents"`
 }
@@ -259,6 +260,10 @@ func (logstash *Logstash) gatherPluginsStats(
 			return err
 		}
 		accumulator.AddFields("logstash_plugins", flattener.Fields, pluginTags)
+		if plugin.Failures != nil {
+			failuresFields := map[string]interface{}{"failures": *plugin.Failures}
+			accumulator.AddFields("logstash_plugins", failuresFields, pluginTags)
+		}
 		/*
 			The elasticsearch output produces additional stats around
 			bulk requests and document writes (that are elasticsearch specific).

plugins/inputs/logstash/logstash_test.go

@@ -335,6 +335,24 @@ func Test_Logstash6GatherPipelinesStats(test *testing.T) {
 		},
 	)
 
+	logstash6accPipelinesStats.AssertContainsTaggedFields(
+		test,
+		"logstash_plugins",
+		map[string]interface{}{
+			"failures": int64(2),
+		},
+		map[string]string{
+			"node_id":      string("3044f675-21ce-4335-898a-8408aa678245"),
+			"node_name":    string("node-6-test"),
+			"source":       string("node-6"),
+			"node_version": string("6.4.2"),
+			"pipeline":     string("main"),
+			"plugin_name":  string("date"),
+			"plugin_id":    string("d079424bb6b7b8c7c61d9c5e0ddae445e92fa9ffa2e8690b0a669f7c690542f0"),
+			"plugin_type":  string("filter"),
+		},
+	)
+
 	logstash6accPipelinesStats.AssertContainsTaggedFields(
 		test,
 		"logstash_plugins",