[Heartbeat] Capture HTTP Response Bodies

heartbeat/docs/fields.asciidoc

@@ -3617,6 +3617,18 @@ alias to: url.full
 
 --
 
+
+
+*`http.response.body.hash`*::
++
+--
+Hash of the response body. Can be used to group responses with identical hashes
+
+
+type: keyword
+
+--
+
 [float]
 === rtt
 

heartbeat/monitors/active/http/_meta/fields.yml

@@ -13,6 +13,16 @@
           migration: true
           description: >
             Service url used by monitor.
+        - name: response
+          type: group
+          fields:
+            - name: body
+              type: group
+              fields:
+                - name: hash
+                  type: keyword
+                  description: >
+                    Hash of the response body. Can be used to group responses with identical hashes
         - name: rtt
           type: group
           description: >

heartbeat/monitors/active/http/config.go

@@ -31,10 +31,12 @@ import (
 )
 
 type Config struct {
-	URLs         []string      `config:"urls" validate:"required"`
-	ProxyURL     string        `config:"proxy_url"`
-	Timeout      time.Duration `config:"timeout"`
-	MaxRedirects int           `config:"max_redirects"`
+	URLs                 []string       `config:"urls" validate:"required"`
+	ProxyURL             string         `config:"proxy_url"`
+	Timeout              time.Duration  `config:"timeout"`
+	MaxRedirects         int            `config:"max_redirects"`
+	Response             responseConfig `config:"response"`
+	TruncateResponseBody string         `config:"truncate_response_body"`
 
 	Mode monitors.IPSettings `config:",inline"`
 
@@ -49,6 +51,11 @@ type Config struct {
 	Check checkConfig `config:"check"`
 }
 
+type responseConfig struct {
+	IncludeBody         string `config:"include_body"`
+	IncludeBodyMaxBytes int    `config:"include_body_max_bytes"`
+}
+
 type checkConfig struct {
 	Request  requestParameters  `config:"request"`
 	Response responseParameters `config:"response"`
@@ -87,7 +94,11 @@ type compressionConfig struct {
 var defaultConfig = Config{
 	Timeout:      16 * time.Second,
 	MaxRedirects: 10,
-	Mode:         monitors.DefaultIPSettings,
+	Response: responseConfig{
+		IncludeBody:         "on_error",
+		IncludeBodyMaxBytes: 2048,
+	},
+	Mode: monitors.DefaultIPSettings,
 	Check: checkConfig{
 		Request: requestParameters{
 			Method:      "GET",
@@ -103,6 +114,20 @@ var defaultConfig = Config{
 	},
 }
 
+func (r *responseConfig) Validate() error {
+	switch strings.ToLower(r.IncludeBody) {
+	case "always", "on_error", "never":
+	default:
+		return fmt.Errorf("unknown option for `include_body`: '%s', please use one of 'always', 'on_error', 'never'", r.IncludeBody)
+	}
+
+	if r.IncludeBodyMaxBytes < 0 {
+		return fmt.Errorf("include_body_max_bytes must be a positive integer, got %d", r.IncludeBodyMaxBytes)
+	}
+
+	return nil
+}
+
 func (r *requestParameters) Validate() error {
 	switch strings.ToUpper(r.Method) {
 	case "HEAD", "GET", "POST":

heartbeat/monitors/active/http/http_test.go

@@ -113,6 +113,13 @@ func respondingHTTPChecks(url string, statusCode int) validator.Validator {
 	)
 }
 
+func respondingHTTPBodyChecks(body string) validator.Validator {
+	return lookslike.MustCompile(map[string]interface{}{
+		"http.response.body.content": body,
+		"http.response.body.bytes":   int64(len(body)),
+	})
+}
+
 var upStatuses = []int{
 	// 1xx
 	http.StatusContinue,
@@ -224,6 +231,7 @@ func TestDownStatuses(t *testing.T) {
 					hbtest.SummaryChecks(0, 1),
 					respondingHTTPChecks(server.URL, status),
 					hbtest.ErrorChecks(fmt.Sprintf("%d", status), "validate"),
+					respondingHTTPBodyChecks("hello, world!"),
 				)),
 				event.Fields,
 			)

heartbeat/monitors/active/http/respbody.go

@@ -0,0 +1,132 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package http
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"net/http"
+	"unicode/utf8"
+
+	"github.com/elastic/beats/heartbeat/eventext"
+	"github.com/elastic/beats/libbeat/common"
+
+	"github.com/elastic/beats/heartbeat/reason"
+	"github.com/elastic/beats/libbeat/beat"
+)
+
+func handleRespBody(event *beat.Event, resp *http.Response, responseConfig responseConfig, errReason reason.Reason) error {
+	defer resp.Body.Close()
+
+	sampleMaxBytes := responseConfig.IncludeBodyMaxBytes
+
+	includeSample := responseConfig.IncludeBody == "always" || (responseConfig.IncludeBody == "on_error" && errReason != nil)
+
+	// No need to return any actual body bytes if we'll discard them anyway. This should save on allocation
+	if !includeSample {
+		sampleMaxBytes = 0
+	}
+
+	sampleStr, bodyBytes, bodyHash, err := readResp(resp, sampleMaxBytes)
+	if err != nil {
+		return err
+	}
+
+	if includeSample {
+		addRespBodyFields(event, sampleStr, bodyBytes, bodyHash)
+	}
+
+	return nil
+}
+
+func addRespBodyFields(event *beat.Event, sampleStr string, bodyBytes int64, bodyHash string) {
+	body := common.MapStr{"bytes": bodyBytes}
+	if sampleStr != "" {
+		body["content"] = sampleStr
+	}
+	if bodyHash != "" {
+		body["hash"] = bodyHash
+	}
+
+	eventext.MergeEventFields(event, common.MapStr{"http": common.MapStr{
+		"response": common.MapStr{
+			"body": body,
+		},
+	}})
+}
+
+// readResp reads the first sampleSize bytes from the httpResponse,
+// then closes the body (which closes the connection). It doesn't return any errors
+// but does log them. During an error case the return values will be (nil, -1).
+// The maxBytes params controls how many bytes will be returned in a string, not how many will be read.
+// We always read the full response here since we want to time downloading the full thing.
+// This may return a nil body if the response is not valid UTF-8
+func readResp(resp *http.Response, maxSampleBytes int) (bodySample string, bodySize int64, hashStr string, err error) {
+	if resp == nil {
+		return "", -1, "", fmt.Errorf("cannot readResp of nil HTTP response")
+	}
+
+	respSize, bodySample, hash, err := readPrefixAndHash(resp.Body, maxSampleBytes)
+
+	return bodySample, respSize, hash, err
+}
+
+func readPrefixAndHash(body io.ReadCloser, maxPrefixSize int) (respSize int64, prefix string, hashStr string, err error) {
+	hash := sha256.New()
+	// Function to lazily get the body of the response
+	rawBuf := make([]byte, 1024)
+
+	// Buffer to hold the prefix output along with tracking info
+	prefixBuf := make([]byte, maxPrefixSize)
+	prefixRemainingBytes := maxPrefixSize
+	prefixWriteOffset := 0
+	for {
+		readSize, readErr := body.Read(rawBuf)
+
+		respSize += int64(readSize)
+		hash.Write(rawBuf[:readSize])
+
+		if prefixRemainingBytes > 0 {
+			if readSize >= prefixRemainingBytes {
+				copy(prefixBuf[prefixWriteOffset:maxPrefixSize], rawBuf[:prefixRemainingBytes])
+				prefixWriteOffset += prefixRemainingBytes
+				prefixRemainingBytes = 0
+			} else {
+				copy(prefixBuf[prefixWriteOffset:prefixWriteOffset+readSize], rawBuf[:readSize])
+				prefixWriteOffset += readSize
+				prefixRemainingBytes -= readSize
+			}
+		}
+
+		if readErr == io.EOF {
+			break
+		}
+
+		if readErr != nil {
+			return 0, "", "", readErr
+		}
+	}
+
+	// We discard the body if it is not valid UTF-8
+	if utf8.Valid(prefixBuf[:prefixWriteOffset]) {
+		prefix = string(prefixBuf[:prefixWriteOffset])
+	}
+	return respSize, prefix, hex.EncodeToString(hash.Sum(nil)), nil
+}