Cherry-pick #23039 to 7.x: [Heartbeat] Make mime type response resolution always on

heartbeat/heartbeat.yml

@@ -130,9 +130,7 @@ processors:
         #name: us-east-1a
         # Lat, Lon "
         #location: "37.926868, -78.024902"
-  - detect_mime_type:
-      field: http.response.body.content
-      target: http.response.mime_type
+
 
 # ================================== Logging ===================================
 

heartbeat/monitors/active/http/http_test.go

@@ -110,9 +110,9 @@ func urlChecks(urlStr string) validator.Validator {
 	})
 }
 
-func respondingHTTPChecks(url string, statusCode int) validator.Validator {
+func respondingHTTPChecks(url, mimeType string, statusCode int) validator.Validator {
 	return lookslike.Compose(
-		minimalRespondingHTTPChecks(url, statusCode),
+		minimalRespondingHTTPChecks(url, mimeType, statusCode),
 		respondingHTTPStatusAndTimingChecks(statusCode),
 		respondingHTTPHeaderChecks(),
 	)
@@ -131,12 +131,13 @@ func respondingHTTPStatusAndTimingChecks(statusCode int) validator.Validator {
 	})
 }
 
-func minimalRespondingHTTPChecks(url string, statusCode int) validator.Validator {
+func minimalRespondingHTTPChecks(url, mimeType string, statusCode int) validator.Validator {
 	return lookslike.Compose(
 		urlChecks(url),
 		httpBodyChecks(),
 		lookslike.MustCompile(map[string]interface{}{
 			"http": map[string]interface{}{
+				"response.mime_type":   mimeType,
 				"response.status_code": statusCode,
 				"rtt.total.us":         isdef.IsDuration,
 			},
@@ -259,7 +260,7 @@ func TestUpStatuses(t *testing.T) {
 						hbtest.BaseChecks("127.0.0.1", "up", "http"),
 						hbtest.RespondingTCPChecks(),
 						hbtest.SummaryChecks(1, 0),
-						respondingHTTPChecks(server.URL, status),
+						respondingHTTPChecks(server.URL, "text/plain; charset=utf-8", status),
 					)),
 					event.Fields,
 				)
@@ -276,7 +277,7 @@ func TestHeadersDisabled(t *testing.T) {
 			hbtest.BaseChecks("127.0.0.1", "up", "http"),
 			hbtest.RespondingTCPChecks(),
 			hbtest.SummaryChecks(1, 0),
-			respondingHTTPChecks(server.URL, 200),
+			respondingHTTPChecks(server.URL, "text/plain; charset=utf-8", 200),
 		)),
 		event.Fields,
 	)
@@ -294,7 +295,7 @@ func TestDownStatuses(t *testing.T) {
 					hbtest.BaseChecks("127.0.0.1", "down", "http"),
 					hbtest.RespondingTCPChecks(),
 					hbtest.SummaryChecks(0, 1),
-					respondingHTTPChecks(server.URL, status),
+					respondingHTTPChecks(server.URL, "text/plain; charset=utf-8", status),
 					hbtest.ErrorChecks(fmt.Sprintf("%d", status), "validate"),
 					respondingHTTPBodyChecks("hello, world!"),
 				)),
@@ -333,7 +334,7 @@ func TestLargeResponse(t *testing.T) {
 			hbtest.BaseChecks("127.0.0.1", "up", "http"),
 			hbtest.RespondingTCPChecks(),
 			hbtest.SummaryChecks(1, 0),
-			respondingHTTPChecks(server.URL, 200),
+			respondingHTTPChecks(server.URL, "text/plain; charset=utf-8", 200),
 		)),
 		event.Fields,
 	)
@@ -394,7 +395,7 @@ func runHTTPSServerCheck(
 			hbtest.RespondingTCPChecks(),
 			hbtest.TLSChecks(0, 0, cert),
 			hbtest.SummaryChecks(1, 0),
-			respondingHTTPChecks(server.URL, http.StatusOK),
+			respondingHTTPChecks(server.URL, "text/plain; charset=utf-8", http.StatusOK),
 		)),
 		event.Fields,
 	)
@@ -549,7 +550,7 @@ func TestRedirect(t *testing.T) {
 			lookslike.Strict(lookslike.Compose(
 				hbtest.BaseChecks("", "up", "http"),
 				hbtest.SummaryChecks(1, 0),
-				minimalRespondingHTTPChecks(testURL, 200),
+				minimalRespondingHTTPChecks(testURL, "text/plain; charset=utf-8", 200),
 				respondingHTTPHeaderChecks(),
 				lookslike.MustCompile(map[string]interface{}{
 					// For redirects that are followed we shouldn't record this header because there's no sensible
@@ -595,7 +596,7 @@ func TestNoHeaders(t *testing.T) {
 			hbtest.SummaryChecks(1, 0),
 			hbtest.RespondingTCPChecks(),
 			respondingHTTPStatusAndTimingChecks(200),
-			minimalRespondingHTTPChecks(server.URL, 200),
+			minimalRespondingHTTPChecks(server.URL, "text/plain; charset=utf-8", 200),
 			lookslike.MustCompile(map[string]interface{}{
 				"http.response.headers": isdef.KeyMissing,
 			}),

heartbeat/monitors/active/http/respbody.go

@@ -28,30 +28,37 @@ import (
 
 	"github.com/elastic/beats/v7/heartbeat/reason"
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/mime"
 )
 
-// maxBufferBodyBytes sets a hard limit on how much we're willing to buffer for any reason internally.
-// since we must buffer the whole body for body validators this is effectively a cap on that.
-// 100MiB out to be enough for everybody.
-const maxBufferBodyBytes = 100 * units.MiB
+const (
+	// maxBufferBodyBytes sets a hard limit on how much we're willing to buffer for any reason internally.
+	// since we must buffer the whole body for body validators this is effectively a cap on that.
+	// 100MiB out to be enough for everybody.
+	maxBufferBodyBytes = 100 * units.MiB
+	// minBufferBytes is the lower bound on how much of the body content we actually read. In order to do
+	// do mime type detection of the response body, we always need a small read buffer.
+	minBufferBodyBytes = 128
+)
 
-func processBody(resp *http.Response, config responseConfig, validator multiValidator) (common.MapStr, reason.Reason) {
+func processBody(resp *http.Response, config responseConfig, validator multiValidator) (common.MapStr, string, reason.Reason) {
 	// Determine how much of the body to actually buffer in memory
 	var bufferBodyBytes int
 	if validator.wantsBody() {
 		bufferBodyBytes = maxBufferBodyBytes
 	} else if config.IncludeBody == "always" || config.IncludeBody == "on_error" {
 		// If the user has asked for bodies to be recorded we only need to buffer that much
 		bufferBodyBytes = config.IncludeBodyMaxBytes
-	} else {
-		// Otherwise, we buffer nothing
-		bufferBodyBytes = 0
+	}
+
+	if bufferBodyBytes < minBufferBodyBytes {
+		bufferBodyBytes = minBufferBodyBytes
 	}
 
 	respBody, bodyLenBytes, bodyHash, respErr := readBody(resp, bufferBodyBytes)
 	// If we encounter an error while reading the body just fail early
 	if respErr != nil {
-		return nil, reason.IOFailed(respErr)
+		return nil, "", reason.IOFailed(respErr)
 	}
 
 	// Run any validations
@@ -77,7 +84,7 @@ func processBody(resp *http.Response, config responseConfig, validator multiVali
 		bodyFields["content"] = respBody[0:sampleNumBytes]
 	}
 
-	return bodyFields, errReason
+	return bodyFields, mime.Detect(respBody), errReason
 }
 
 // readBody reads the first sampleSize bytes from the httpResponse,

heartbeat/monitors/active/http/respbody_test.go

@@ -44,6 +44,7 @@ func Test_handleRespBody(t *testing.T) {
 
 	type args struct {
 		resp           *http.Response
+		mimeType       string
 		responseConfig responseConfig
 		validator      multiValidator
 	}
@@ -57,6 +58,7 @@ func Test_handleRespBody(t *testing.T) {
 			"on_error with error",
 			args{
 				simpleHTTPResponse("hello"),
+				"text/plain; charset=utf-8",
 				responseConfig{IncludeBody: "on_error", IncludeBodyMaxBytes: 3},
 				failingComboValidator,
 			},
@@ -67,6 +69,7 @@ func Test_handleRespBody(t *testing.T) {
 			"on_error with success",
 			args{
 				simpleHTTPResponse("hello"),
+				"text/plain; charset=utf-8",
 				responseConfig{IncludeBody: "on_error", IncludeBodyMaxBytes: 3},
 				matchingComboValidator,
 			},
@@ -77,6 +80,7 @@ func Test_handleRespBody(t *testing.T) {
 			"always with error",
 			args{
 				simpleHTTPResponse("hello"),
+				"text/plain; charset=utf-8",
 				responseConfig{IncludeBody: "always", IncludeBodyMaxBytes: 3},
 				failingComboValidator,
 			},
@@ -87,6 +91,7 @@ func Test_handleRespBody(t *testing.T) {
 			"always with success",
 			args{
 				simpleHTTPResponse("hello"),
+				"text/plain; charset=utf-8",
 				responseConfig{IncludeBody: "always", IncludeBodyMaxBytes: 3},
 				matchingComboValidator,
 			},
@@ -97,6 +102,7 @@ func Test_handleRespBody(t *testing.T) {
 			"never with error",
 			args{
 				simpleHTTPResponse("hello"),
+				"text/plain; charset=utf-8",
 				responseConfig{IncludeBody: "never", IncludeBodyMaxBytes: 3},
 				failingComboValidator,
 			},
@@ -107,6 +113,7 @@ func Test_handleRespBody(t *testing.T) {
 			"never with success",
 			args{
 				simpleHTTPResponse("hello"),
+				"text/plain; charset=utf-8",
 				responseConfig{IncludeBody: "never", IncludeBodyMaxBytes: 3},
 				matchingComboValidator,
 			},
@@ -117,10 +124,13 @@ func Test_handleRespBody(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			fields, err := processBody(tt.args.resp, tt.args.responseConfig, tt.args.validator)
+			fields, mimeType, err := processBody(tt.args.resp, tt.args.responseConfig, tt.args.validator)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("handleRespBody() error = %v, wantErr %v", err, tt.wantErr)
 			}
+			if mimeType != tt.args.mimeType {
+				t.Errorf("invalid mime type - got: '%v' - want: '%v'", mimeType, tt.args.mimeType)
+			}
 
 			bodyMatch := map[string]interface{}{
 				"hash":  "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824",

heartbeat/monitors/active/http/task.go

@@ -249,13 +249,17 @@ func execPing(
 		return start, time.Now(), errReason
 	}
 
-	bodyFields, errReason := processBody(resp, responseConfig, validator)
+	bodyFields, mimeType, errReason := processBody(resp, responseConfig, validator)
 
 	responseFields := common.MapStr{
 		"status_code": resp.StatusCode,
 		"body":        bodyFields,
 	}
 
+	if mimeType != "" {
+		responseFields["mime_type"] = mimeType
+	}
+
 	if responseConfig.IncludeHeaders {
 		headerFields := common.MapStr{}
 		for canonicalHeaderKey, vals := range resp.Header {

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

@@ -85,11 +85,6 @@ fields:
   {% endfor -%}
 {% endif %}
 
-processors:
-  - detect_mime_type:
-      field: http.response.body.content
-      target: http.response.mime_type
-
 #================================ Queue =====================================
 
 queue.mem: