[Heartbeat] Correctly store HTTP bodies with validation (elastic#14223)…

… (elastic#14310)

Currently, when using an HTTP body validator (either regexp or JSON) will break the storage of HTTP bodies with response.include_body (introduced in elastic#13022).

The root cause is that both validation and reading the body for inclusion in the event share the same ReadCloser provided by *http.Response.

This patch looks at both validation and body settings to determine how much of the body to read, reads that much, then passes that to the validation and body inclusion code as a []byte.

Resolves elastic#13751

(cherry picked from commit 46643b0)

CHANGELOG.next.asciidoc

@@ -21,6 +21,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Heartbeat*
 
+- JSON/Regex checks against HTTP bodies will only consider the first 100MiB of the HTTP body to prevent excessive memory usage. {pull}14223[pull]
 
 *Journalbeat*
 
@@ -53,6 +54,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Heartbeat*
 
+- Fix storage of HTTP bodies to work when JSON/Regex body checks are enabled. {pull}14223[14223]
 
 *Journalbeat*
 

heartbeat/docs/heartbeat-options.asciidoc

@@ -500,7 +500,8 @@ Under `check.response`, specify these options:
 
 *`status`*:: The expected status code. 4xx and 5xx codes are considered `down` by default. Other codes are considered `up`.
 *`headers`*:: The required response headers.
-*`body`*:: A list of regular expressions to match the the body output. Only a single expression needs to match.
+*`body`*:: A list of regular expressions to match the the body output. Only a single expression needs to match. HTTP response
+bodies of up to 100MiB are supported.
 
 Example configuration:
 This monitor examines the
@@ -524,7 +525,8 @@ response body for the strings `saved` or `Saved`
       - saved
 -------------------------------------------------------------------------------
 
-*`json`*:: A list of <<conditions,condition>> expressions executed against the body when parsed as JSON.
+*`json`*:: A list of <<conditions,condition>> expressions executed against the body when parsed as JSON. Body sizes
+must be less than or equal to 100 MiB.
 
 The following configuration shows how to check the response when the body
 contains JSON:

heartbeat/monitors/active/http/check.go

@@ -27,68 +27,82 @@ import (
 
 	pkgerrors "github.com/pkg/errors"
 
+	"github.com/elastic/beats/heartbeat/reason"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/jsontransform"
 	"github.com/elastic/beats/libbeat/common/match"
 	"github.com/elastic/beats/libbeat/conditions"
 )
 
-type RespCheck func(*http.Response) error
+// multiValidator combines multiple validations of each type into a single easy to use object.
+type multiValidator struct {
+	respValidators []respValidator
+	bodyValidators []bodyValidator
+}
+
+func (rv multiValidator) wantsBody() bool {
+	return len(rv.bodyValidators) > 0
+}
+
+func (rv multiValidator) validate(resp *http.Response, body string) reason.Reason {
+	for _, respValidator := range rv.respValidators {
+		if err := respValidator(resp); err != nil {
+			return reason.ValidateFailed(err)
+		}
+	}
+
+	for _, bodyValidator := range rv.bodyValidators {
+		if err := bodyValidator(resp, body); err != nil {
+			return reason.ValidateFailed(err)
+		}
+	}
+
+	return nil
+}
+
+// respValidator is used for validating using only the non-body fields of the *http.Response.
+// Accessing the body of the response in such a validator should not be done due, use bodyValidator
+// for those purposes instead.
+type respValidator func(*http.Response) error
+
+// bodyValidator lets you validate a stringified version of the body along with other metadata in
+// *http.Response.
+type bodyValidator func(*http.Response, string) error
 
 var (
 	errBodyMismatch = errors.New("body mismatch")
 )
 
-func makeValidateResponse(config *responseParameters) (RespCheck, error) {
-	var checks []RespCheck
+func makeValidateResponse(config *responseParameters) (multiValidator, error) {
+	var respValidators []respValidator
+	var bodyValidators []bodyValidator
 
 	if config.Status > 0 {
-		checks = append(checks, checkStatus(config.Status))
+		respValidators = append(respValidators, checkStatus(config.Status))
 	} else {
-		checks = append(checks, checkStatusOK)
+		respValidators = append(respValidators, checkStatusOK)
 	}
 
 	if len(config.RecvHeaders) > 0 {
-		checks = append(checks, checkHeaders(config.RecvHeaders))
+		respValidators = append(respValidators, checkHeaders(config.RecvHeaders))
 	}
 
 	if len(config.RecvBody) > 0 {
-		checks = append(checks, checkBody(config.RecvBody))
+		bodyValidators = append(bodyValidators, checkBody(config.RecvBody))
 	}
 
 	if len(config.RecvJSON) > 0 {
 		jsonChecks, err := checkJSON(config.RecvJSON)
 		if err != nil {
-			return nil, err
+			return multiValidator{}, err
 		}
-		checks = append(checks, jsonChecks)
+		bodyValidators = append(bodyValidators, jsonChecks)
 	}
 
-	return checkAll(checks...), nil
+	return multiValidator{respValidators, bodyValidators}, nil
 }
 
-func checkOK(_ *http.Response) error { return nil }
-
-// TODO: collect all errors into on error message.
-func checkAll(checks ...RespCheck) RespCheck {
-	switch len(checks) {
-	case 0:
-		return checkOK
-	case 1:
-		return checks[0]
-	}
-
-	return func(r *http.Response) error {
-		for _, check := range checks {
-			if err := check(r); err != nil {
-				return err
-			}
-		}
-		return nil
-	}
-}
-
-func checkStatus(status uint16) RespCheck {
+func checkStatus(status uint16) respValidator {
 	return func(r *http.Response) error {
 		if r.StatusCode == int(status) {
 			return nil
@@ -104,7 +118,7 @@ func checkStatusOK(r *http.Response) error {
 	return nil
 }
 
-func checkHeaders(headers map[string]string) RespCheck {
+func checkHeaders(headers map[string]string) respValidator {
 	return func(r *http.Response) error {
 		for k, v := range headers {
 			value := r.Header.Get(k)
@@ -116,22 +130,18 @@ func checkHeaders(headers map[string]string) RespCheck {
 	}
 }
 
-func checkBody(body []match.Matcher) RespCheck {
-	return func(r *http.Response) error {
-		content, err := ioutil.ReadAll(r.Body)
-		if err != nil {
-			return err
-		}
-		for _, m := range body {
-			if m.Match(content) {
+func checkBody(matcher []match.Matcher) bodyValidator {
+	return func(r *http.Response, body string) error {
+		for _, m := range matcher {
+			if m.MatchString(body) {
 				return nil
 			}
 		}
 		return errBodyMismatch
 	}
 }
 
-func checkJSON(checks []*jsonResponseCheck) (RespCheck, error) {
+func checkJSON(checks []*jsonResponseCheck) (bodyValidator, error) {
 	type compiledCheck struct {
 		description string
 		condition   conditions.Condition
@@ -147,9 +157,9 @@ func checkJSON(checks []*jsonResponseCheck) (RespCheck, error) {
 		compiledChecks = append(compiledChecks, compiledCheck{check.Description, cond})
 	}
 
-	return func(r *http.Response) error {
+	return func(r *http.Response, body string) error {
 		decoded := &common.MapStr{}
-		decoder := json.NewDecoder(r.Body)
+		decoder := json.NewDecoder(strings.NewReader(body))
 		decoder.UseNumber()
 		err := decoder.Decode(decoded)
 

heartbeat/monitors/active/http/check_test.go

@@ -19,15 +19,15 @@ package http
 
 import (
 	"fmt"
+	"io/ioutil"
 	"log"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 
-	"github.com/elastic/beats/libbeat/common"
-
 	"github.com/stretchr/testify/require"
 
+	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/match"
 	"github.com/elastic/beats/libbeat/conditions"
 )
@@ -118,7 +118,9 @@ func TestCheckBody(t *testing.T) {
 			for _, pattern := range test.patterns {
 				patterns = append(patterns, match.MustCompile(pattern))
 			}
-			check := checkBody(patterns)(res)
+			body, err := ioutil.ReadAll(res.Body)
+			require.NoError(t, err)
+			check := checkBody(patterns)(res, string(body))
 
 			if result := (check == nil); result != test.result {
 				if test.result {
@@ -183,7 +185,9 @@ func TestCheckJson(t *testing.T) {
 
 			checker, err := checkJSON([]*jsonResponseCheck{{test.condDesc, test.condConf}})
 			require.NoError(t, err)
-			checkRes := checker(res)
+			body, err := ioutil.ReadAll(res.Body)
+			require.NoError(t, err)
+			checkRes := checker(res, string(body))
 
 			if result := checkRes == nil; result != test.result {
 				if test.result {
@@ -249,7 +253,9 @@ func TestCheckJsonWithIntegerComparison(t *testing.T) {
 
 			checker, err := checkJSON([]*jsonResponseCheck{{test.condDesc, test.condConf}})
 			require.NoError(t, err)
-			checkRes := checker(res)
+			body, err := ioutil.ReadAll(res.Body)
+			require.NoError(t, err)
+			checkRes := checker(res, string(body))
 
 			if result := checkRes == nil; result != test.result {
 				if test.result {

heartbeat/monitors/active/http/http_test.go

@@ -41,8 +41,6 @@ import (
 	btesting "github.com/elastic/beats/libbeat/testing"
 	"github.com/elastic/go-lookslike"
 	"github.com/elastic/go-lookslike/isdef"
-	"github.com/elastic/go-lookslike/llpath"
-	"github.com/elastic/go-lookslike/llresult"
 	"github.com/elastic/go-lookslike/testslike"
 	"github.com/elastic/go-lookslike/validator"
 )
@@ -112,18 +110,7 @@ func respondingHTTPChecks(url string, statusCode int) validator.Validator {
 				"response.status_code": statusCode,
 				"response.body.hash":   isdef.IsString,
 				// TODO add this isdef to lookslike in a robust way
-				"response.body.bytes": isdef.Is("an int64 greater than 0", func(path llpath.Path, v interface{}) *llresult.Results {
-					raw, ok := v.(int64)
-					if !ok {
-						return llresult.SimpleResult(path, false, "%s is not an int64", reflect.TypeOf(v))
-					}
-					if raw >= 0 {
-						return llresult.ValidResult(path)
-					}
-
-					return llresult.SimpleResult(path, false, "value %v not >= 0 ", raw)
-
-				}),
+				"response.body.bytes":    isdef.IsIntGt(-1),
 				"rtt.content.us":         isdef.IsDuration,
 				"rtt.response_header.us": isdef.IsDuration,
 				"rtt.total.us":           isdef.IsDuration,
@@ -134,10 +121,30 @@ func respondingHTTPChecks(url string, statusCode int) validator.Validator {
 	)
 }
 
+func minimalRespondingHTTPChecks(url string, statusCode int) validator.Validator {
+	return lookslike.Compose(
+		httpBaseChecks(url),
+		httpBodyChecks(),
+		lookslike.MustCompile(map[string]interface{}{
+			"http": map[string]interface{}{
+				"response.status_code": statusCode,
+				"rtt.total.us":         isdef.IsDuration,
+			},
+		}),
+	)
+}
+
+func httpBodyChecks() validator.Validator {
+	return lookslike.MustCompile(map[string]interface{}{
+		"http.response.body.bytes": isdef.IsIntGt(-1),
+		"http.response.body.hash":  isdef.IsString,
+	})
+}
+
 func respondingHTTPBodyChecks(body string) validator.Validator {
 	return lookslike.MustCompile(map[string]interface{}{
 		"http.response.body.content": body,
-		"http.response.body.bytes":   int64(len(body)),
+		"http.response.body.bytes":   len(body),
 	})
 }