Merge pull request #70 from NETWAYS/feature/check-reload-config

Add check for pipeline configuration reload

Author: martialblog

README.md

@@ -86,6 +86,28 @@ Flags:
   -h, --help                          help for pipeline
 ```
 
+### Pipeline Reload
+
+Checks the status of Logstash pipelines configuration reload.
+
+```bash
+Usage:
+  check_logstash pipeline reload [flags]
+
+Examples:
+
+	$ check_logstash pipeline reload
+	OK - Configuration successfully reloaded
+	 \_[OK] Configuration successfully reloaded for pipeline Foobar for on 2021-01-01T02:07:14Z
+
+	$ check_logstash pipeline reload --pipeline Example
+	CRITICAL - Configuration reload failed
+	 \_[CRITICAL] Configuration reload for pipeline Example failed on 2021-01-01T02:07:14Z
+
+Flags:
+  -P, --pipeline string               Pipeline Name (default "/")
+  -h, --help                          help for pipeline
+```
 
 ## License
 

cmd/pipeline.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"time"
 )
 
 // To store the CLI parameters
@@ -161,9 +162,105 @@ var pipelineCmd = &cobra.Command{
 	},
 }
 
+var pipelineReloadCmd = &cobra.Command{
+	Use:   "reload",
+	Short: "Checks the reload configuration status of the Logstash Pipelines",
+	Long:  `Checks the reload configuration status of the Logstash Pipelines`,
+	Example: `
+	$ check_logstash pipeline reload
+	OK - Configuration successfully reloaded
+	 \_[OK] Configuration successfully reloaded for pipeline Foobar for on 2021-01-01T02:07:14Z
+
+	$ check_logstash pipeline reload --pipeline Example
+	CRITICAL - Configuration reload failed
+	 \_[CRITICAL] Configuration reload for pipeline Example failed on 2021-01-01T02:07:14Z`,
+	Run: func(cmd *cobra.Command, args []string) {
+		var (
+			output string
+			rc     int
+			pp     logstash.Pipeline
+		)
+
+		// 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 reload configuration status for each pipeline
+		var summary strings.Builder
+
+		for name, pipe := range pp.Pipelines {
+			// Check Reload Timestamp
+			if pipe.Reloads.LastSuccessTime != "" {
+				// We could do the parsing during the unmarshall
+				lastSuccessReload, errSu := time.Parse(time.RFC3339, pipe.Reloads.LastSuccessTime)
+				lastFailureReload, errFa := time.Parse(time.RFC3339, pipe.Reloads.LastFailureTime)
+
+				if errSu != nil || errFa != nil {
+					states = append(states, check.Unknown)
+					summary.WriteString(fmt.Sprintf("[UNKNOWN] Configuration reload for pipeline %s unknown;", name))
+					summary.WriteString("\n  \\_")
+					continue
+				}
+
+				summary.WriteString("\n  \\_")
+				if lastFailureReload.After(lastSuccessReload) {
+					states = append(states, check.Critical)
+					summary.WriteString(fmt.Sprintf("[CRITICAL] Configuration reload for pipeline %s failed on %s;", name, lastFailureReload))
+				} else {
+					states = append(states, check.OK)
+					summary.WriteString(fmt.Sprintf("[OK] Configuration successfully reloaded for pipeline %s for on %s;", name, lastSuccessReload))
+				}
+			}
+		}
+
+		// Validate the various subchecks and use the worst state as return code
+		switch result.WorstState(states...) {
+		case 0:
+			rc = check.OK
+			output = "Configuration successfully reloaded"
+		case 1:
+			rc = check.Warning
+			output = "Configuration reload may not be successful"
+		case 2:
+			rc = check.Critical
+			output = "Configuration reload failed"
+		default:
+			rc = check.Unknown
+			output = "Configuration reload status unknown"
+		}
+
+		check.ExitRaw(rc, output, summary.String())
+	},
+}
+
 func init() {
 	rootCmd.AddCommand(pipelineCmd)
 
+	pipelineReloadCmd.Flags().StringVarP(&cliPipelineConfig.PipelineName, "pipeline", "P", "/",
+		"Pipeline Name")
+
+	pipelineCmd.AddCommand(pipelineReloadCmd)
+
 	fs := pipelineCmd.Flags()
 
 	// Default is / since we use this value for the URL Join

cmd/pipeline_test.go

@@ -117,6 +117,51 @@ func TestPipelineCmd_Logstash8(t *testing.T) {
 			args:     []string{"run", "../main.go", "pipeline", "--inflight-events-warn", "200", "--inflight-events-crit", "500"},
 			expected: "OK - Inflight events",
 		},
+		{
+			name: "pipeline-reload-no-success",
+			server: httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				w.Write([]byte(`{"host":"localhost","version":"8.6","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":{"localhost-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":"","last_error":null,"last_failure_timestamp":"2020-10-11T01:10:10.11Z","failures":0},"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", "reload"},
+			expected: "UNKNOWN - Configuration reload status unknown",
+		},
+		{
+			name: "pipeline-reload-not-timestamp",
+			server: httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				w.Write([]byte(`{"host":"localhost","version":"8.6","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":{"localhost-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":"not a timestamp","last_error":null,"last_failure_timestamp":"no time for you","failures":0},"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", "reload"},
+			expected: "[UNKNOWN] Configuration reload for pipeline",
+		},
+		{
+			name: "pipeline-reload-failed",
+			server: httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				w.Write([]byte(`{"host":"localhost","version":"8.6","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":{"localhost-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":"2020-10-11T01:10:10.11Z","last_error":null,"last_failure_timestamp":"2021-10-11T01:10:10.11Z","failures":0},"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", "reload"},
+			expected: "[CRITICAL] Configuration reload for pipeline localhost-input",
+		},
+		{
+			name: "pipeline-reload-ok",
+			server: httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				w.Write([]byte(`{"host":"localhost","version":"8.6","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":{"localhost-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":"2020-10-11T01:10:10.11Z","last_error":null,"last_failure_timestamp":"2019-10-11T01:10:10.11Z","failures":0},"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", "reload"},
+			expected: "[OK] Configuration successfully reloaded for pipeline localhost-input",
+		},
+		{
+			name: "pipeline-reload-missing",
+			server: httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusNotFound)
+				w.Write([]byte(`{"path": "/_node/stats/pipelines/foo","status": 404,"error": {"message": "Not Found"}}`))
+			})),
+			args:     []string{"run", "../main.go", "pipeline", "reload", "--pipeline", "foo"},
+			expected: "UNKNOWN - Could not get",
+		},
 	}
 
 	for _, test := range tests {

internal/logstash/api.go

@@ -13,8 +13,10 @@ type Pipeline struct {
 	Host      string `json:"host"`
 	Pipelines map[string]struct {
 		Reloads struct {
-			Successes int `json:"successes"`
-			Failures  int `json:"failures"`
+			LastSuccessTime string `json:"last_success_timestamp"`
+			LastFailureTime string `json:"last_failure_timestamp"`
+			Successes       int    `json:"successes"`
+			Failures        int    `json:"failures"`
 		} `json:"reloads"`
 		Queue struct {
 			Type                string `json:"type"`