Adds isPipelineHealthy function

collectors/nodestats/pipeline_subcollector.go

@@ -15,6 +15,7 @@ import (
 // pipelines of a logstash node.
 // The collector is created once for each pipeline of the node.
 type PipelineSubcollector struct {
+	Up                      *prometheus.Desc
 	EventsOut               *prometheus.Desc
 	EventsFiltered          *prometheus.Desc
 	EventsIn                *prometheus.Desc
@@ -34,6 +35,7 @@ type PipelineSubcollector struct {
 func NewPipelineSubcollector() *PipelineSubcollector {
 	descHelper := prometheus_helper.SimpleDescHelper{Namespace: namespace, Subsystem: fmt.Sprintf("%s_pipeline", subsystem)}
 	return &PipelineSubcollector{
+		Up:                      descHelper.NewDescWithHelpAndLabel("up", "Whether the pipeline is up or not.", "pipeline_id"),
 		EventsOut:               descHelper.NewDescWithHelpAndLabel("events_out", "Number of events that have been processed by this pipeline.", "pipeline_id"),
 		EventsFiltered:          descHelper.NewDescWithHelpAndLabel("events_filtered", "Number of events that have been filtered out by this pipeline.", "pipeline_id"),
 		EventsIn:                descHelper.NewDescWithHelpAndLabel("events_in", "Number of events that have been inputted into this pipeline.", "pipeline_id"),
@@ -79,3 +81,21 @@ func (collector *PipelineSubcollector) Collect(pipeStats *responses.SinglePipeli
 	collectingEnd := time.Now()
 	log.Printf("collected pipeline stats for pipeline %s in %s", pipelineID, collectingEnd.Sub(collectingStart))
 }
+
+func (collector *PipelineSubcollector) isPipelineHealthy(pipeReloadStats responses.PipelineReloadResponse) float64 {
+	// 1. If both timestamps are nil, the pipeline is healthy
+	if pipeReloadStats.LastSuccessTimestamp == nil && pipeReloadStats.LastFailureTimestamp == nil {
+		return 1
+	// 2. If last_failure_timestamp is set and last success timestamp is nil, the pipeline is unhealthy
+	} else if pipeReloadStats.LastFailureTimestamp != nil && pipeReloadStats.LastSuccessTimestamp == nil {
+		return 0
+	// 3. If last_success_timestamp < last_failure_timestamp, the pipeline is unhealthy
+	} else if pipeReloadStats.LastSuccessTimestamp.Before(*pipeReloadStats.LastFailureTimestamp) {
+		return 0
+	// 4. If last_success_timestamp > last_failure_timestamp, the pipeline is healthy
+	} else if pipeReloadStats.LastSuccessTimestamp.After(*pipeReloadStats.LastFailureTimestamp) {
+		return 1
+	}
+	// Missing field, likely due to version incompatibility - lacking information, assume healthy
+	return 1
+}