Merge pull request #72 from NETWAYS/feature/flow-metrics

martialblog · web-flow · commit cfbd5e920a91 · 2023-06-26T13:24:04.000+02:00
Add Flow Metrics
diff --git a/README.md b/README.md
@@ -62,7 +62,9 @@ Flags:
 
 ### Pipeline
 
-Checks the status of Logstash pipelines.
+Determines the health of Logstash pipelines via "inflight events". These events are calculated as such: `inflight events = events.In - events.Out`
+
+Hint: Use the queue backpressure for Logstash 8.
 
 ```bash
 Usage:
@@ -86,6 +88,34 @@ Flags:
   -h, --help                          help for pipeline
 ```
 
+### Pipeline Flow Metrics
+
+Checks the status of a Logstash pipeline's flow metrics (currently queue backpressure).
+
+Hint: Requires Logstash 8.5.0
+
+```bash
+
+Usage:
+  check_logstash pipeline flow [flags]
+
+Examples:
+
+	$ check_logstash pipeline flow --warning 5 --critical 10
+	OK - Flow metrics alright
+	 \_[OK] queue_backpressure_example:0.34;
+
+	$ check_logstash pipeline flow --pipeline example --warning 5 --critical 10
+	CRITICAL - Flow metrics alright
+	 \_[CRITICAL] queue_backpressure_example:11.23;
+
+Flags:
+  -c, --critical string   Critical threshold for queue Backpressure
+  -h, --help              help for flow
+  -P, --pipeline string   Pipeline Name (default "/")
+  -w, --warning string    Warning threshold for queue Backpressure
+```
+
 ### Pipeline Reload
 
 Checks the status of Logstash pipelines configuration reload.
diff --git a/cmd/pipeline.go b/cmd/pipeline.go
@@ -16,15 +16,15 @@ import (
 
 // To store the CLI parameters
 type PipelineConfig struct {
-	PipelineName       string
-	InflightEventsWarn string
-	InflightEventsCrit string
+	PipelineName string
+	Warning      string
+	Critical     string
 }
 
 // To store the parsed CLI parameters
 type PipelineThreshold struct {
-	inflightEventsWarn *check.Threshold
-	inflightEventsCrit *check.Threshold
+	Warning  *check.Threshold
+	Critical *check.Threshold
 }
 
 var cliPipelineConfig PipelineConfig
@@ -33,19 +33,19 @@ func parsePipeThresholds(config PipelineConfig) (PipelineThreshold, error) {
 	// Parses the CLI parameters
 	var t PipelineThreshold
 
-	inflightEventsWarn, err := check.ParseThreshold(config.InflightEventsWarn)
+	warn, err := check.ParseThreshold(config.Warning)
 	if err != nil {
 		return t, err
 	}
 
-	t.inflightEventsWarn = inflightEventsWarn
+	t.Warning = warn
 
-	inflightEventsCrit, err := check.ParseThreshold(config.InflightEventsCrit)
+	crit, err := check.ParseThreshold(config.Critical)
 	if err != nil {
 		return t, err
 	}
 
-	t.inflightEventsCrit = inflightEventsCrit
+	t.Critical = crit
 
 	return t, nil
 }
@@ -109,10 +109,10 @@ var pipelineCmd = &cobra.Command{
 			inflightEvents := pipe.Events.In - pipe.Events.Out
 
 			summary.WriteString("\n \\_")
-			if thresholds.inflightEventsCrit.DoesViolate(float64(inflightEvents)) {
+			if thresholds.Critical.DoesViolate(float64(inflightEvents)) {
 				states = append(states, check.Critical)
 				summary.WriteString(fmt.Sprintf("[CRITICAL] inflight_events_%s:%d;", name, inflightEvents))
-			} else if thresholds.inflightEventsWarn.DoesViolate(float64(inflightEvents)) {
+			} else if thresholds.Warning.DoesViolate(float64(inflightEvents)) {
 				states = append(states, check.Warning)
 				summary.WriteString(fmt.Sprintf("[WARNING] inflight_events_%s:%d;", name, inflightEvents))
 			} else {
@@ -131,8 +131,8 @@ var pipelineCmd = &cobra.Command{
 				Value: pipe.Events.Out})
 			perfList.Add(&perfdata.Perfdata{
 				Label: fmt.Sprintf("inflight_events_%s", name),
-				Warn:  thresholds.inflightEventsWarn,
-				Crit:  thresholds.inflightEventsCrit,
+				Warn:  thresholds.Warning,
+				Crit:  thresholds.Critical,
 				Value: inflightEvents})
 			perfList.Add(&perfdata.Perfdata{
 				Label: fmt.Sprintf("pipelines.%s.reloads.failures", name),
@@ -253,13 +253,128 @@ var pipelineReloadCmd = &cobra.Command{
 	},
 }
 
+var pipelineFlowCmd = &cobra.Command{
+	Use:   "flow",
+	Short: "Checks the flow metrics of the Logstash Pipelines",
+	Long:  `Checks the flow metrics of the Logstash Pipelines`,
+	Example: `
+	$ check_logstash pipeline flow --warning 5 --critical 10
+	OK - Flow metrics alright
+	 \_[OK] queue_backpressure_example:0.34;
+
+	$ check_logstash pipeline flow --pipeline example --warning 5 --critical 10
+	CRITICAL - Flow metrics not alright
+	 \_[CRITICAL] queue_backpressure_example:11.23;`,
+	Run: func(cmd *cobra.Command, args []string) {
+		var (
+			output     string
+			rc         int
+			thresholds PipelineThreshold
+			pp         logstash.Pipeline
+			perfList   perfdata.PerfdataList
+		)
+
+		// Parse the thresholds into a central var since we need them later
+		thresholds, err := parsePipeThresholds(cliPipelineConfig)
+		if err != nil {
+			check.ExitError(err)
+		}
+
+		// Creating an client and connecting to the API
+		c := cliConfig.NewClient()
+		// localhost:9600/_node/stats/pipelines/ will return all Pipelines
+		// localhost:9600/_node/stats/pipelines/foo will return the foo Pipeline
+		u, _ := url.JoinPath(c.Url, "/_node/stats/pipelines", cliPipelineConfig.PipelineName)
+		resp, err := c.Client.Get(u)
+
+		if err != nil {
+			check.ExitError(err)
+		}
+
+		if resp.StatusCode != http.StatusOK {
+			check.ExitError(fmt.Errorf("Could not get %s - Error: %d", u, resp.StatusCode))
+		}
+
+		defer resp.Body.Close()
+		err = json.NewDecoder(resp.Body).Decode(&pp)
+
+		if err != nil {
+			check.ExitError(err)
+		}
+
+		states := make([]int, 0, len(pp.Pipelines))
+
+		// Check the flow metrics for each pipeline
+		var summary strings.Builder
+
+		for name, pipe := range pp.Pipelines {
+			summary.WriteString("\n \\_")
+			if thresholds.Critical.DoesViolate(pipe.Flow.QueueBackpressure.Current) {
+				states = append(states, check.Critical)
+				summary.WriteString(fmt.Sprintf("[CRITICAL] queue_backpressure_%s:%.2f;", name, pipe.Flow.QueueBackpressure.Current))
+			} else if thresholds.Warning.DoesViolate(pipe.Flow.QueueBackpressure.Current) {
+				states = append(states, check.Warning)
+				summary.WriteString(fmt.Sprintf("[WARNING] queue_backpressure_%s:%.2f;", name, pipe.Flow.QueueBackpressure.Current))
+			} else {
+				states = append(states, check.OK)
+				summary.WriteString(fmt.Sprintf("[OK] queue_backpressure_%s:%.2f;", name, pipe.Flow.QueueBackpressure.Current))
+			}
+
+			// Generate perfdata for each event
+			perfList.Add(&perfdata.Perfdata{
+				Label: fmt.Sprintf("pipelines.queue_backpressure_%s", name),
+				Warn:  thresholds.Warning,
+				Crit:  thresholds.Critical,
+				Value: pipe.Flow.QueueBackpressure.Current})
+			perfList.Add(&perfdata.Perfdata{
+				Label: fmt.Sprintf("pipelines.%s.output_throughput", name),
+				Value: pipe.Flow.OutputThroughput.Current})
+			perfList.Add(&perfdata.Perfdata{
+				Label: fmt.Sprintf("pipelines.%s.input_throughput", name),
+				Value: pipe.Flow.InputThroughput.Current})
+			perfList.Add(&perfdata.Perfdata{
+				Label: fmt.Sprintf("pipelines.%s.filter_throughput", name),
+				Value: pipe.Flow.FilterThroughput.Current})
+		}
+
+		// Validate the various subchecks and use the worst state as return code
+		switch result.WorstState(states...) {
+		case 0:
+			rc = check.OK
+			output = "Flow metrics alright"
+		case 1:
+			rc = check.Warning
+			output = "Flow metrics may not be alright"
+		case 2:
+			rc = check.Critical
+			output = "Flow metrics not alright"
+		default:
+			rc = check.Unknown
+			output = "Flow metrics status unknown"
+		}
+
+		check.ExitRaw(rc, output, summary.String(), "|", perfList.String())
+	},
+}
+
 func init() {
 	rootCmd.AddCommand(pipelineCmd)
 
 	pipelineReloadCmd.Flags().StringVarP(&cliPipelineConfig.PipelineName, "pipeline", "P", "/",
 		"Pipeline Name")
 
+	pipelineFlowCmd.Flags().StringVarP(&cliPipelineConfig.PipelineName, "pipeline", "P", "/",
+		"Pipeline Name")
+	pipelineFlowCmd.Flags().StringVarP(&cliPipelineConfig.Warning, "warning", "w", "",
+		"Warning threshold for queue Backpressure")
+	pipelineFlowCmd.Flags().StringVarP(&cliPipelineConfig.Critical, "critical", "c", "",
+		"Critical threshold for queue Backpressure")
+
+	_ = pipelineFlowCmd.MarkFlagRequired("warning")
+	_ = pipelineFlowCmd.MarkFlagRequired("critical")
+
 	pipelineCmd.AddCommand(pipelineReloadCmd)
+	pipelineCmd.AddCommand(pipelineFlowCmd)
 
 	fs := pipelineCmd.Flags()
 
@@ -268,9 +383,9 @@ func init() {
 	fs.StringVarP(&cliPipelineConfig.PipelineName, "pipeline", "P", "/",
 		"Pipeline Name")
 
-	fs.StringVar(&cliPipelineConfig.InflightEventsWarn, "inflight-events-warn", "",
+	fs.StringVar(&cliPipelineConfig.Warning, "inflight-events-warn", "",
 		"Warning threshold for inflight events to be a warning result. Use min:max for a range.")
-	fs.StringVar(&cliPipelineConfig.InflightEventsCrit, "inflight-events-crit", "",
+	fs.StringVar(&cliPipelineConfig.Critical, "inflight-events-crit", "",
 		"Critical threshold for inflight events to be a critical result. Use min:max for a range.")
 
 	_ = pipelineCmd.MarkFlagRequired("inflight-events-warn")
diff --git a/cmd/pipeline_test.go b/cmd/pipeline_test.go
@@ -162,6 +162,24 @@ func TestPipelineCmd_Logstash8(t *testing.T) {
 			args:     []string{"run", "../main.go", "pipeline", "reload", "--pipeline", "foo"},
 			expected: "UNKNOWN - Could not get",
 		},
+		{
+			name: "pipeline-flow-ok",
+			server: httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				w.Write([]byte(`{"host":"foobar","version":"8.7.1","http_address":"127.0.0.1:9600","id":"4","name":"test","ephemeral_id":"5","status":"green","snapshot":false,"pipeline":{"workers":2,"batch_size":125,"batch_delay":50},"pipelines":{"ansible-input":{"flow":{"queue_backpressure":{"current":12.34,"last_1_minute":0,"lifetime":2.503e-05},"output_throughput":{"current":0,"last_1_minute":0.344,"lifetime":0.7051},"input_throughput":{"current":10,"last_1_minute":0.5734,"lifetime":1.089},"worker_concurrency":{"current":0.0001815,"last_1_minute":0.0009501,"lifetime":0.003384},"filter_throughput":{"current":0,"last_1_minute":0.5734,"lifetime":1.089}},"events":{"filtered":0,"duration_in_millis":0,"queue_push_duration_in_millis":0,"out":50,"in":100},"queue":{"type":"memory","events_count":0,"queue_size_in_bytes":0,"max_queue_size_in_bytes":0},"hash":"f","ephemeral_id":"f"}}}`))
+			})),
+			args:     []string{"run", "../main.go", "pipeline", "flow", "--warning", "15", "--critical", "20"},
+			expected: "[OK] queue_backpressure_ansible-input:12.34; | pipelines.queue_backpressure_ansible-input=12.34;15;20 pipelines.ansible-input.output_throughput=0 pipelines.ansible-input.input_throughput=10 pipelines.ansible-input.filter_throughput=0",
+		},
+		{
+			name: "pipeline-flow-critical",
+			server: httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				w.Write([]byte(`{"host":"foobar","version":"8.7.1","http_address":"127.0.0.1:9600","id":"4","name":"test","ephemeral_id":"5","status":"green","snapshot":false,"pipeline":{"workers":2,"batch_size":125,"batch_delay":50},"pipelines":{"ansible-input":{"flow":{"queue_backpressure":{"current":10,"last_1_minute":0,"lifetime":2.503e-05},"output_throughput":{"current":0,"last_1_minute":0.344,"lifetime":0.7051},"input_throughput":{"current":10,"last_1_minute":0.5734,"lifetime":1.089},"worker_concurrency":{"current":0.0001815,"last_1_minute":0.0009501,"lifetime":0.003384},"filter_throughput":{"current":0,"last_1_minute":0.5734,"lifetime":1.089}},"events":{"filtered":0,"duration_in_millis":0,"queue_push_duration_in_millis":0,"out":50,"in":100},"queue":{"type":"memory","events_count":0,"queue_size_in_bytes":0,"max_queue_size_in_bytes":0},"hash":"f","ephemeral_id":"f"}}}`))
+			})),
+			args:     []string{"run", "../main.go", "pipeline", "flow", "--warning", "1", "--critical", "2"},
+			expected: "CRITICAL - Flow metrics not alright",
+		},
 	}
 
 	for _, test := range tests {
diff --git a/internal/logstash/api.go b/internal/logstash/api.go
@@ -18,6 +18,12 @@ type Pipeline struct {
 			Successes       int    `json:"successes"`
 			Failures        int    `json:"failures"`
 		} `json:"reloads"`
+		Flow struct {
+			QueueBackpressure FlowMetric `json:"queue_backpressure"`
+			OutputThroughput  FlowMetric `json:"output_throughput"`
+			InputThroughput   FlowMetric `json:"input_throughput"`
+			FilterThroughput  FlowMetric `json:"filter_throughput"`
+		} `json:"flow"`
 		Queue struct {
 			Type                string `json:"type"`
 			EventsCount         int    `json:"events_count"`
@@ -34,6 +40,12 @@ type Pipeline struct {
 	} `json:"pipelines"`
 }
 
+type FlowMetric struct {
+	Current     float64 `json:"current"`
+	Last1Minute float64 `json:"last_1_minute"`
+	Lifetime    float64 `json:"lifetime"`
+}
+
 type Process struct {
 	MaxFileDescriptors  float64 `json:"max_file_descriptors"`
 	OpenFileDescriptors float64 `json:"open_file_descriptors"`
diff --git a/internal/logstash/api_test.go b/internal/logstash/api_test.go
@@ -5,6 +5,26 @@ import (
 	"testing"
 )
 
+func TestUmarshallPipelineFlow(t *testing.T) {
+
+	j := `{"host":"foobar","version":"8.7.1","http_address":"127.0.0.1:9600","id":"4","name":"test","ephemeral_id":"5","status":"green","snapshot":false,"pipeline":{"workers":2,"batch_size":125,"batch_delay":50},"pipelines":{"ansible-input":{"flow":{"queue_backpressure":{"current":10,"last_1_minute":0,"lifetime":2.503e-05},"output_throughput":{"current":0,"last_1_minute":0.344,"lifetime":0.7051},"input_throughput":{"current":10,"last_1_minute":0.5734,"lifetime":1.089},"worker_concurrency":{"current":0.0001815,"last_1_minute":0.0009501,"lifetime":0.003384},"filter_throughput":{"current":0,"last_1_minute":0.5734,"lifetime":1.089}},"events":{"filtered":0,"duration_in_millis":0,"queue_push_duration_in_millis":0,"out":50,"in":100},"plugins":{"inputs":[{"id":"b","name":"beats","events":{"queue_push_duration_in_millis":0,"out":0}}],"codecs":[{"id":"plain","name":"plain","decode":{"writes_in":0,"duration_in_millis":0,"out":0},"encode":{"writes_in":0,"duration_in_millis":0}},{"id":"json","name":"json","decode":{"writes_in":0,"duration_in_millis":0,"out":0},"encode":{"writes_in":0,"duration_in_millis":0}}],"filters":[],"outputs":[{"id":"f","name":"redis","events":{"duration_in_millis":18,"out":50,"in":100}}]},"reloads":{"successes":0,"last_success_timestamp":null,"last_error":null,"last_failure_timestamp":null,"failures":0},"queue":{"type":"memory","events_count":0,"queue_size_in_bytes":0,"max_queue_size_in_bytes":0},"hash":"f","ephemeral_id":"f"}}}`
+
+	var pl Pipeline
+	err := json.Unmarshal([]byte(j), &pl)
+
+	if err != nil {
+		t.Error(err)
+	}
+
+	if pl.Pipelines["ansible-input"].Flow.QueueBackpressure.Current != 10 {
+		t.Error("\nActual: ", pl.Pipelines["ansible-input"].Flow.QueueBackpressure.Current, "\nExpected: ", "10")
+	}
+
+	if pl.Pipelines["ansible-input"].Flow.InputThroughput.Current != 10 {
+		t.Error("\nActual: ", pl.Pipelines["ansible-input"].Flow.InputThroughput.Current, "\nExpected: ", "10")
+	}
+}
+
 func TestUmarshallPipeline(t *testing.T) {
 
 	j := `{"host":"foobar","version":"7.17.8","http_address":"127.0.0.1:9600","id":"4","name":"test","ephemeral_id":"5","status":"green","snapshot":false,"pipeline":{"workers":2,"batch_size":125,"batch_delay":50},"pipelines":{"ansible-input":{"events":{"filtered":0,"duration_in_millis":0,"queue_push_duration_in_millis":0,"out":50,"in":100},"plugins":{"inputs":[{"id":"b","name":"beats","events":{"queue_push_duration_in_millis":0,"out":0}}],"codecs":[{"id":"plain","name":"plain","decode":{"writes_in":0,"duration_in_millis":0,"out":0},"encode":{"writes_in":0,"duration_in_millis":0}},{"id":"json","name":"json","decode":{"writes_in":0,"duration_in_millis":0,"out":0},"encode":{"writes_in":0,"duration_in_millis":0}}],"filters":[],"outputs":[{"id":"f","name":"redis","events":{"duration_in_millis":18,"out":50,"in":100}}]},"reloads":{"successes":0,"last_success_timestamp":null,"last_error":null,"last_failure_timestamp":null,"failures":0},"queue":{"type":"memory","events_count":0,"queue_size_in_bytes":0,"max_queue_size_in_bytes":0},"hash":"f","ephemeral_id":"f"}}}`
