Adding bulk requests errors metric

collector/nodestats_api.go

@@ -38,6 +38,9 @@ type Pipeline struct {
 				In  int `json:"in"`
 				Out int `json:"out"`
 			} `json:"events"`
+			BulkRequests struct {
+				Errors int `json:"with_errors"`
+			} `json:"bulk_requests"`
 			Name string `json:"name"`
 		} `json:"outputs"`
 	} `json:"plugins"`

collector/nodestats_collector.go

@@ -49,6 +49,8 @@ type NodeStatsCollector struct {
 	PipelineQueueMaxUnreadEvents *prometheus.Desc
 
 	PipelineDeadLetterQueueSizeInBytes *prometheus.Desc
+
+	PipelineBulkRequestsWithErrors *prometheus.Desc
 }
 
 // NewNodeStatsCollector function
@@ -233,6 +235,13 @@ func NewNodeStatsCollector(logstashEndpoint string) (Collector, error) {
 			nil,
 		),
 
+		PipelineBulkRequestsWithErrors: prometheus.NewDesc(
+			prometheus.BuildFQName(Namespace, subsystem, "bulk_requests_with_errors"),
+			"bulk_requests_with_errors",
+			[]string{"pipeline", "plugin", "plugin_id", "plugin_type"},
+			nil,
+		),
+
 		PipelinePluginMatches: prometheus.NewDesc(
 			prometheus.BuildFQName(Namespace, subsystem, "plugin_matches_total"),
 			"plugin_matches",
@@ -623,6 +632,15 @@ func (c *NodeStatsCollector) collect(ch chan<- prometheus.Metric) (*prometheus.D
 				plugin.ID,
 				"output",
 			)
+			ch <- prometheus.MustNewConstMetric(
+				c.PipelineBulkRequestsWithErrors,
+				prometheus.CounterValue,
+				float64(plugin.BulkRequests.Errors),
+				pipelineID,
+				plugin.Name,
+				plugin.ID,
+				"output",
+			)
 		}
 
 		if pipeline.Queue.Type != "memory" {

collector/nodestats_collector_test.go

@@ -109,6 +109,12 @@ var noQueueJSON = []byte(`
             "in": 0,
             "out": 0
           },
+	  "bulk_requests": {
+	    "with_errors": 1111,
+	    "responses": {
+	      "200": 1111 
+	    }
+	  },
           "name": "stdout"
         }
       ]