Read whole body on each request (#8894)

We did not read the entire HTTP body unless the user had a regex match declared. We should always do this as part of our HTTP checks, otherwise timing is incorrect.

CHANGELOG.asciidoc

@@ -94,6 +94,7 @@ https://github.com/elastic/beats/compare/v6.4.0...master[Check the HEAD diff]
 *Heartbeat*
 
 - Fixed bug where HTTP responses with larger bodies would incorrectly report connection errors. {pull}8660[8660]
+- Heartbeat now always downloads the entire body of HTTP endpoints, even if no checks against the body content are declared. This fixes an issue where timing metrics would be incorrect in scenarios where the body wasn't used since the connection would be closed soon after the headers were sent, but before the entire body was. {pull}8894[8894]
 
 *Journalbeat*
 

heartbeat/monitors/active/http/task.go

@@ -21,6 +21,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -260,12 +261,15 @@ func execRequest(client *http.Client, req *http.Request, validator func(*http.Re
 	}
 
 	err = validator(resp)
-	end = time.Now()
 	if err != nil {
-		return start, end, resp, reason.ValidateFailed(err)
+		return start, time.Now(), resp, reason.ValidateFailed(err)
 	}
 
-	return start, end, resp, nil
+	// Read the entirety of the body. Otherwise, the stats for the check
+	// don't include download time.
+	io.Copy(ioutil.Discard, resp.Body)
+
+	return start, time.Now(), resp, nil
 }
 
 func makeEvent(rtt time.Duration, resp *http.Response) common.MapStr {

heartbeat/tests/system/heartbeat.py

@@ -8,6 +8,7 @@
     __file__), '../../../libbeat/tests/system'))
 
 from beat.beat import TestCase
+from time import sleep
 
 
 class BaseTest(TestCase):
@@ -19,12 +20,15 @@ def setUpClass(self):
             os.path.join(os.path.dirname(__file__), "../../"))
         super(BaseTest, self).setUpClass()
 
-    def start_server(self, content, status_code):
+    def start_server(self, content, status_code, **kwargs):
         class HTTPHandler(BaseHTTPServer.BaseHTTPRequestHandler):
             def do_GET(self):
                 self.send_response(status_code)
                 self.send_header('Content-Type', 'application/json')
                 self.end_headers()
+                if "write_delay" in kwargs:
+                    sleep(float(kwargs["write_delay"]))
+
                 self.wfile.write(content)
 
         server = BaseHTTPServer.HTTPServer(('localhost', 0), HTTPHandler)

heartbeat/tests/system/test_monitor.py

@@ -17,12 +17,8 @@ def test_http(self, status_code):
         status_code = int(status_code)
         server = self.start_server("hello world", status_code)
 
-        self.render_config_template(
-            monitors=[{
-                "type": "http",
-                "urls": ["http://localhost:{}".format(server.server_port)],
-            }],
-        )
+        self.render_http_config(
+            ["http://localhost:{}".format(server.server_port)])
 
         proc = self.start_beat()
         self.wait_until(lambda: self.log_contains("heartbeat is running"))
@@ -41,6 +37,29 @@ def test_http(self, status_code):
             raise SkipTest
         self.assert_fields_are_documented(output[0])
 
+    def test_http_delayed(self):
+        """
+        Ensure that the HTTP monitor consumes the whole body.
+        We do this by ensuring that a slow HTTP body write's time is reflected
+        in the beats metrics.
+        """
+        try:
+            delay = 1.0
+            server = self.start_server("sloooow body", 200, write_delay=delay)
+
+            self.render_http_config(
+                ["http://localhost:{}".format(server.server_port)])
+
+            try:
+                proc = self.start_beat()
+                self.wait_until(lambda: self.output_has(lines=1))
+                nose.tools.assert_greater_equal(
+                    self.last_output_line()['http.rtt.total.us'], delay)
+            finally:
+                proc.check_kill_and_wait()
+        finally:
+            server.shutdown()
+
     @parameterized.expand([
         ("up", '{"foo": {"baz": "bar"}}'),
         ("down", '{"foo": "unexpected"}'),
@@ -115,3 +134,11 @@ def test_tcp(self, url, status):
             self.assert_fields_are_documented(output[0])
         finally:
             server.shutdown()
+
+    def render_http_config(self, urls):
+        self.render_config_template(
+            monitors=[{
+                "type": "http",
+                "urls": urls,
+            }]
+        )