JSON Checks for Heartbeat HTTP Monitors

CHANGELOG.asciidoc

@@ -156,6 +156,7 @@ https://github.com/elastic/beats/compare/v6.4.0...master[Check the HEAD diff]
 *Journalbeat*
 
 - Add journalbeat. {pull}8703[8703]
+- Add the ability to check against JSON HTTP bodies with conditions. {pull}8667[8667]
 
 *Metricbeat*
 

heartbeat/_meta/beat.reference.yml

@@ -208,6 +208,14 @@ heartbeat.monitors:
     # Required response contents.
     #body:
 
+    # Parses the body as JSON, then checks against the given condition expression
+    #json:
+    #- description: Explanation of what the check does
+    #  condition:
+    #    equals:
+    #      myField: expectedValue
+
+
   # NOTE: THIS FEATURE IS DEPRECATED AND WILL BE REMOVED IN A FUTURE RELEASE
   # Configure file json file to be watched for changes to the monitor:
   #watch.poll_file:

heartbeat/docs/heartbeat-options.asciidoc

@@ -444,6 +444,7 @@ Under `check.response`, specify these options:
 it's set to 0, any status code other than 404 is accepted.
 *`headers`*:: The required response headers.
 *`body`*:: A list of regular expressions to match the the body output. Only a single expression needs to match.
+*`json`*:: A list of <<conditions,condition>> expressions executed against the body when parsed as JSON.
 
 The following configuration shows how to check the response when the body
 contains JSON:
@@ -459,7 +460,11 @@ contains JSON:
       'X-API-Key': '12345-mykey-67890'
   check.response:
     status: 200
-    body: '{"status": "ok"}'
+    json:
+      - description: check status
+        condition:
+          equals:
+            status: ok
 -------------------------------------------------------------------------------
 
 The following configuration shows how to check the response for multiple regex

heartbeat/heartbeat.reference.yml

@@ -208,6 +208,14 @@ heartbeat.monitors:
     # Required response contents.
     #body:
 
+    # Parses the body as JSON, then checks against the given condition expression
+    #json:
+    #- description: Explanation of what the check does
+    #  condition:
+    #    equals:
+    #      myField: expectedValue
+
+
   # NOTE: THIS FEATURE IS DEPRECATED AND WILL BE REMOVED IN A FUTURE RELEASE
   # Configure file json file to be watched for changes to the monitor:
   #watch.poll_file:

heartbeat/monitors/active/http/check.go

@@ -18,12 +18,18 @@
 package http
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
+	"strings"
 
+	pkgerrors "github.com/pkg/errors"
+
+	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/match"
+	"github.com/elastic/beats/libbeat/conditions"
 )
 
 type RespCheck func(*http.Response) error
@@ -32,7 +38,7 @@ var (
 	errBodyMismatch = errors.New("body mismatch")
 )
 
-func makeValidateResponse(config *responseParameters) RespCheck {
+func makeValidateResponse(config *responseParameters) (RespCheck, error) {
 	var checks []RespCheck
 
 	if config.Status > 0 {
@@ -49,7 +55,15 @@ func makeValidateResponse(config *responseParameters) RespCheck {
 		checks = append(checks, checkBody(config.RecvBody))
 	}
 
-	return checkAll(checks...)
+	if len(config.RecvJSON) > 0 {
+		jsonChecks, err := checkJSON(config.RecvJSON)
+		if err != nil {
+			return nil, err
+		}
+		checks = append(checks, jsonChecks)
+	}
+
+	return checkAll(checks...), nil
 }
 
 func checkOK(_ *http.Response) error { return nil }
@@ -115,3 +129,49 @@ func checkBody(body []match.Matcher) RespCheck {
 		return errBodyMismatch
 	}
 }
+
+func checkJSON(checks []*jsonResponseCheck) (RespCheck, error) {
+	type compiledCheck struct {
+		description string
+		condition   conditions.Condition
+	}
+
+	var compiledChecks []compiledCheck
+
+	for _, check := range checks {
+		cond, err := conditions.NewCondition(check.Condition)
+		if err != nil {
+			return nil, err
+		}
+		compiledChecks = append(compiledChecks, compiledCheck{check.Description, cond})
+	}
+
+	return func(r *http.Response) error {
+		decoded := &common.MapStr{}
+		err := json.NewDecoder(r.Body).Decode(decoded)
+
+		if err != nil {
+			body, _ := ioutil.ReadAll(r.Body)
+			return pkgerrors.Wrapf(err, "could not parse JSON for body check with condition. Source: %s", body)
+		}
+
+		var errorDescs []string
+		for _, compiledCheck := range compiledChecks {
+			ok := compiledCheck.condition.Check(decoded)
+			if !ok {
+				errorDescs = append(errorDescs, compiledCheck.description)
+			}
+		}
+
+		if len(errorDescs) > 0 {
+			return fmt.Errorf(
+				"JSON body did not match %d conditions '%s' for monitor. Received JSON %+v",
+				len(errorDescs),
+				strings.Join(errorDescs, ","),
+				decoded,
+			)
+		}
+
+		return nil
+	}, nil
+}

heartbeat/monitors/active/http/check_test.go

@@ -24,7 +24,12 @@ import (
 	"net/http/httptest"
 	"testing"
 
+	"github.com/elastic/beats/libbeat/common"
+
+	"github.com/stretchr/testify/require"
+
 	"github.com/elastic/beats/libbeat/common/match"
+	"github.com/elastic/beats/libbeat/conditions"
 )
 
 func TestCheckBody(t *testing.T) {
@@ -125,3 +130,69 @@ func TestCheckBody(t *testing.T) {
 		})
 	}
 }
+
+func TestCheckJson(t *testing.T) {
+	fooBazEqualsBar := common.MustNewConfigFrom(map[string]interface{}{"equals": map[string]interface{}{"foo": map[string]interface{}{"baz": "bar"}}})
+	fooBazEqualsBarConf := &conditions.Config{}
+	err := fooBazEqualsBar.Unpack(fooBazEqualsBarConf)
+	require.NoError(t, err)
+
+	fooBazEqualsBarDesc := "foo.baz equals bar"
+
+	var tests = []struct {
+		description string
+		body        string
+		condDesc    string
+		condConf    *conditions.Config
+		result      bool
+	}{
+		{
+			"positive match",
+			"{\"foo\": {\"baz\": \"bar\"}}",
+			fooBazEqualsBarDesc,
+			fooBazEqualsBarConf,
+			true,
+		},
+		{
+			"Negative match",
+			"{\"foo\": 123}",
+			fooBazEqualsBarDesc,
+			fooBazEqualsBarConf,
+			false,
+		},
+		{
+			"unparseable",
+			`notjson`,
+			fooBazEqualsBarDesc,
+			fooBazEqualsBarConf,
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.description, func(t *testing.T) {
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				fmt.Fprintln(w, test.body)
+			}))
+			defer ts.Close()
+
+			res, err := http.Get(ts.URL)
+			if err != nil {
+				log.Fatal(err)
+			}
+
+			checker, err := checkJSON([]*jsonResponseCheck{{test.condDesc, test.condConf}})
+			require.NoError(t, err)
+			checkRes := checker(res)
+
+			if result := checkRes == nil; result != test.result {
+				if test.result {
+					t.Fatalf("Expected condition: '%s' to match body: %s. got: %s", test.condDesc, test.body, checkRes)
+				} else {
+					t.Fatalf("Did not expect condition: '%s' to match body: %s. got: %s", test.condDesc, test.body, checkRes)
+				}
+			}
+		})
+	}
+
+}

heartbeat/monitors/active/http/config.go

@@ -22,6 +22,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/elastic/beats/libbeat/conditions"
+
 	"github.com/elastic/beats/libbeat/common/match"
 	"github.com/elastic/beats/libbeat/common/transport/tlscommon"
 
@@ -68,9 +70,15 @@ type requestParameters struct {
 
 type responseParameters struct {
 	// expected HTTP response configuration
-	Status      uint16            `config:"status" verify:"min=0, max=699"`
-	RecvHeaders map[string]string `config:"headers"`
-	RecvBody    []match.Matcher   `config:"body"`
+	Status      uint16               `config:"status" verify:"min=0, max=699"`
+	RecvHeaders map[string]string    `config:"headers"`
+	RecvBody    []match.Matcher      `config:"body"`
+	RecvJSON    []*jsonResponseCheck `config:"json"`
+}
+
+type jsonResponseCheck struct {
+	Description string             `config:"description"`
+	Condition   *conditions.Config `config:"condition"`
 }
 
 type compressionConfig struct {
@@ -93,6 +101,7 @@ var defaultConfig = Config{
 			Status:      0,
 			RecvHeaders: nil,
 			RecvBody:    []match.Matcher{},
+			RecvJSON:    nil,
 		},
 	},
 }

heartbeat/monitors/active/http/http.go

@@ -70,7 +70,10 @@ func create(
 		body = buf.Bytes()
 	}
 
-	validator := makeValidateResponse(&config.Check.Response)
+	validator, err := makeValidateResponse(&config.Check.Response)
+	if err != nil {
+		return nil, 0, err
+	}
 
 	jobs = make([]monitors.Job, len(config.URLs))
 

heartbeat/monitors/active/http/simple_transp.go

@@ -184,6 +184,7 @@ func (t *SimpleTransport) readResponse(
 ) (*http.Response, error) {
 	reader := bufio.NewReader(conn)
 	resp, err := http.ReadResponse(reader, req)
+	resp.Body = comboConnReadCloser{conn, resp.Body}
 	if err != nil {
 		return nil, err
 	}

heartbeat/tests/system/config/heartbeat.yml.j2

@@ -29,6 +29,14 @@ heartbeat.monitors:
     {% endfor %}
   {% endif -%}
 
+
+  {%- if monitor.check_response_json is defined %}
+  check.response.json:
+    {%- for check in monitor.check_response_json %}
+    - {{check}}
+    {% endfor %}
+  {% endif -%}
+
   {%- if monitor.fields is defined %}
   {% if monitor.fields_under_root %}fields_under_root: true{% endif %}
   fields:

heartbeat/tests/system/test_monitor.py

@@ -41,6 +41,43 @@ def test_http(self, status_code):
             raise SkipTest
         self.assert_fields_are_documented(output[0])
 
+    @parameterized.expand([
+        ("up", '{"foo": {"baz": "bar"}}'),
+        ("down", '{"foo": "unexpected"}'),
+        ("down", 'notjson'),
+    ])
+    def test_http_json(self, expected_status, body):
+        """
+        Test JSON response checks
+        """
+        server = self.start_server(body, 200)
+        try:
+            self.render_config_template(
+                monitors=[{
+                    "type": "http",
+                    "urls": ["http://localhost:{}".format(server.server_port)],
+                    "check_response_json": [{
+                        "description": "foo equals bar",
+                        "condition": {
+                            "equals": {"foo": {"baz": "bar"}}
+                        }
+                    }]
+                }]
+            )
+
+            try:
+                proc = self.start_beat()
+                self.wait_until(lambda: self.log_contains("heartbeat is running"))
+
+                self.wait_until(
+                    lambda: self.output_has(lines=1))
+            finally:
+                proc.check_kill_and_wait()
+
+            self.assert_last_status(expected_status)
+        finally:
+            server.shutdown()
+
     @parameterized.expand([
         (lambda server: "localhost:{}".format(server.server_port), "up"),
         # This IP is reserved in IPv4