Skip to content

Commit

Permalink
[Heartbeat] Capture HTTP Response Bodies (elastic#13022)
Browse files Browse the repository at this point in the history
This PR adds optional support for capturing HTTP response bodies in heartbeat events.

By default it sets the body only on responses that return errors, but can optionally be set to never report the body or report it on all checks.
  • Loading branch information
andrewvc authored Aug 27, 2019
1 parent 1a7c61d commit 1e3e65c
Show file tree
Hide file tree
Showing 10 changed files with 513 additions and 44 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.next.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -286,6 +286,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
*Heartbeat*

- Enable `add_observer_metadata` processor in default config. {pull}11394[11394]
- Record HTTP body metadata and optionally contents in `http.response.body.*` fields. {pull}13022[13022]

*Journalbeat*

Expand Down
12 changes: 12 additions & 0 deletions heartbeat/docs/fields.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -4371,6 +4371,18 @@ alias to: url.full
--
*`http.response.body.hash`*::
+
--
Hash of the full response body. Can be used to group responses with identical hashes
type: keyword
--
[float]
=== rtt
Expand Down
16 changes: 15 additions & 1 deletion heartbeat/docs/heartbeat-options.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -428,11 +428,25 @@ Example configuration:

Also see <<configuration-ssl>> for a full description of the `ssl` options.

[float]
[[monitor-http-response]]
=== `response`

Controls the indexing of the HTTP response body contents to the `http.response.body.contents` field.

Set `response.include_body` to one of the options listed below.

*`on_error`*:: Include the body if an error is encountered during the check. This is the default.
*`never`*:: Never include the body.
*`always`*:: Always include the body with checks.

Set `response.include_body_max_bytes` to control the maximum size of the stored body contents. Defaults to 1024 bytes.

[float]
[[monitor-http-check]]
==== `check`

An optional `request` to send to the remote host and the expected `response`.
An optional `request` to send to the remote host and the expected `response`.

Example configuration:

Expand Down
2 changes: 1 addition & 1 deletion heartbeat/include/fields.go

Large diffs are not rendered by default.

13 changes: 11 additions & 2 deletions heartbeat/monitors/active/http/_meta/fields.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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 full response body. Can be used to group responses with identical hashes.
- name: rtt
type: group
description: >
Expand Down Expand Up @@ -70,8 +80,7 @@

- name: content.us
type: long
description:
Time required to retrieved the content in micro seconds.
description: Time required to retrieved the content in micro seconds.

- name: total
type: group
Expand Down
34 changes: 29 additions & 5 deletions heartbeat/monitors/active/http/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,10 +31,11 @@ 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"`

Mode monitors.IPSettings `config:",inline"`

Expand All @@ -49,6 +50,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"`
Expand Down Expand Up @@ -87,7 +93,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",
Expand All @@ -103,6 +113,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":
Expand Down
8 changes: 8 additions & 0 deletions heartbeat/monitors/active/http/http_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
)
Expand Down
126 changes: 126 additions & 0 deletions heartbeat/monitors/active/http/respbody.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
// 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
}

evtBodyMap := common.MapStr{
"hash": bodyHash,
"bytes": bodyBytes,
}
if includeSample {
evtBodyMap["content"] = sampleStr
}

eventext.MergeEventFields(event, common.MapStr{
"http": common.MapStr{
"response": common.MapStr{"body": evtBodyMap},
},
})

return nil
}

// 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
}
Loading

0 comments on commit 1e3e65c

Please sign in to comment.