[8.x](backport #41498) x-pack/filebeat/input/http_endpoint: remove raw message retention

CHANGELOG-developer.next.asciidoc

@@ -107,6 +107,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Add the Offset property to libbeat/reader.Message to store the total number of bytes read and discarded before generating the message. This enables inputs to accurately determine how much data has been read up to the message, using Message.Bytes + Message.Offset. {pull}39873[39873] {issue}39653[39653]
 - AWS CloudWatch Metrics record previous endTime to use for next collection period and change log.logger from cloudwatch to aws.cloudwatch. {pull}40870[40870]
 - Fix flaky test in cel and httpjson inputs of filebeat. {issue}40503[40503] {pull}41358[41358]
+- Fix documentation and implementation of raw message handling in Filebeat http_endpoint by removing it. {pull}41498[41498]
 
 ==== Added
 

x-pack/filebeat/docs/inputs/input-http-endpoint.asciidoc

@@ -332,7 +332,7 @@ For example, `["content-type"]` will become `["Content-Type"]` when the filebeat
 [float]
 ==== `preserve_original_event`
 
-This option copies the raw unmodified body of the incoming request to the event.original field as a string before sending the event to Elasticsearch.
+This option includes the JSON representation of the incoming request in the `event.original` field as a string before sending the event to Elasticsearch. The representation may not be a verbatim copy of the original message, but is guaranteed to be an [RFC7493](https://datatracker.ietf.org/doc/html/rfc7493) compliant message.
 
 [float]
 ==== `crc.provider`

x-pack/filebeat/input/http_endpoint/handler.go

@@ -119,7 +119,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		r.Body = io.NopCloser(&buf)
 	}
 
-	objs, _, status, err := httpReadJSON(body, h.program)
+	objs, status, err := httpReadJSON(body, h.program)
 	if err != nil {
 		h.sendAPIErrorResponse(txID, w, r, h.log, status, err)
 		h.metrics.apiErrors.Add(1)
@@ -333,64 +333,65 @@ func (h *handler) publishEvent(obj, headers mapstr.M, acker *batchACKTracker) er
 	return nil
 }
 
-func httpReadJSON(body io.Reader, prg *program) (objs []mapstr.M, rawMessages []json.RawMessage, status int, err error) {
+func httpReadJSON(body io.Reader, prg *program) (objs []mapstr.M, status int, err error) {
 	if body == http.NoBody {
-		return nil, nil, http.StatusNotAcceptable, errBodyEmpty
+		return nil, http.StatusNotAcceptable, errBodyEmpty
 	}
-	obj, rawMessage, err := decodeJSON(body, prg)
+	obj, err := decodeJSON(body, prg)
 	if err != nil {
-		return nil, nil, http.StatusBadRequest, err
+		return nil, http.StatusBadRequest, err
 	}
-	return obj, rawMessage, http.StatusOK, err
+	return obj, http.StatusOK, err
 }
 
-func decodeJSON(body io.Reader, prg *program) (objs []mapstr.M, rawMessages []json.RawMessage, err error) {
+func decodeJSON(body io.Reader, prg *program) (objs []mapstr.M, err error) {
 	decoder := json.NewDecoder(body)
 	for decoder.More() {
 		var raw json.RawMessage
 		if err = decoder.Decode(&raw); err != nil {
 			if err == io.EOF { //nolint:errorlint // This will never be a wrapped error.
 				break
 			}
-			return nil, nil, fmt.Errorf("malformed JSON object at stream position %d: %w", decoder.InputOffset(), err)
+			return nil, fmt.Errorf("malformed JSON object at stream position %d: %w", decoder.InputOffset(), err)
 		}
 
 		var obj interface{}
 		if err = newJSONDecoder(bytes.NewReader(raw)).Decode(&obj); err != nil {
-			return nil, nil, fmt.Errorf("malformed JSON object at stream position %d: %w", decoder.InputOffset(), err)
+			return nil, fmt.Errorf("malformed JSON object at stream position %d: %w", decoder.InputOffset(), err)
 		}
 
 		if prg != nil {
 			obj, err = prg.eval(obj)
 			if err != nil {
-				return nil, nil, err
+				return nil, err
 			}
-			// Re-marshal to ensure the raw bytes agree with the constructed object.
-			raw, err = json.Marshal(obj)
-			if err != nil {
-				return nil, nil, fmt.Errorf("failed to remarshal object: %w", err)
+			if _, ok := obj.([]interface{}); ok {
+				// Re-marshal to ensure the raw bytes agree with the constructed object.
+				// This is only necessary when the program constructs an array return.
+				raw, err = json.Marshal(obj)
+				if err != nil {
+					return nil, fmt.Errorf("failed to remarshal object: %w", err)
+				}
 			}
 		}
 
 		switch v := obj.(type) {
 		case map[string]interface{}:
 			objs = append(objs, v)
-			rawMessages = append(rawMessages, raw)
 		case []interface{}:
-			nobjs, nrawMessages, err := decodeJSONArray(bytes.NewReader(raw))
+			nobjs, err := decodeJSONArray(bytes.NewReader(raw))
 			if err != nil {
-				return nil, nil, fmt.Errorf("recursive error %d: %w", decoder.InputOffset(), err)
+				return nil, fmt.Errorf("recursive error %d: %w", decoder.InputOffset(), err)
 			}
 			objs = append(objs, nobjs...)
-			rawMessages = append(rawMessages, nrawMessages...)
 		default:
-			return nil, nil, fmt.Errorf("%w: %T", errUnsupportedType, v)
+			return nil, fmt.Errorf("%w: %T", errUnsupportedType, v)
 		}
 	}
 	for i := range objs {
 		jsontransform.TransformNumbers(objs[i])
 	}
-	return objs, rawMessages, nil
+	return objs, nil
 }
 
 type program struct {
@@ -506,17 +507,17 @@ func (p *program) eval(obj interface{}) (interface{}, error) {
 	}
 }
 
-func decodeJSONArray(raw *bytes.Reader) (objs []mapstr.M, rawMessages []json.RawMessage, err error) {
+func decodeJSONArray(raw *bytes.Reader) (objs []mapstr.M, err error) {
 	dec := newJSONDecoder(raw)
 	token, err := dec.Token()
 	if err != nil {
 		if err == io.EOF { //nolint:errorlint // This will never be a wrapped error.
-			return nil, nil, nil
+			return nil, nil
 		}
-		return nil, nil, fmt.Errorf("failed reading JSON array: %w", err)
+		return nil, fmt.Errorf("failed reading JSON array: %w", err)
 	}
 	if token != json.Delim('[') {
-		return nil, nil, fmt.Errorf("malformed JSON array, not starting with delimiter [ at position: %d", dec.InputOffset())
+		return nil, fmt.Errorf("malformed JSON array, not starting with delimiter [ at position: %d", dec.InputOffset())
 	}
 
 	for dec.More() {
@@ -525,21 +526,20 @@ func decodeJSONArray(raw *bytes.Reader) (objs []mapstr.M, rawMessages []json.Raw
 			if err == io.EOF { //nolint:errorlint // This will never be a wrapped error.
 				break
 			}
-			return nil, nil, fmt.Errorf("malformed JSON object at stream position %d: %w", dec.InputOffset(), err)
+			return nil, fmt.Errorf("malformed JSON object at stream position %d: %w", dec.InputOffset(), err)
 		}
 
 		var obj interface{}
 		if err := newJSONDecoder(bytes.NewReader(raw)).Decode(&obj); err != nil {
-			return nil, nil, fmt.Errorf("malformed JSON object at stream position %d: %w", dec.InputOffset(), err)
+			return nil, fmt.Errorf("malformed JSON object at stream position %d: %w", dec.InputOffset(), err)
 		}
 
 		m, ok := obj.(map[string]interface{})
 		if ok {
-			rawMessages = append(rawMessages, raw)
 			objs = append(objs, m)
 		}
 	}
-	return objs, rawMessages, nil
+	return objs, nil
 }
 
 func getIncludedHeaders(r *http.Request, headerConf []string) (includedHeaders mapstr.M) {

x-pack/filebeat/input/http_endpoint/handler_test.go

@@ -8,7 +8,6 @@ import (
 	"bytes"
 	"compress/gzip"
 	"context"
-	"encoding/json"
 	"errors"
 	"flag"
 	"io"
@@ -38,13 +37,12 @@ func Test_httpReadJSON(t *testing.T) {
 	log := logp.NewLogger("http_endpoint_test")
 
 	tests := []struct {
-		name           string
-		body           string
-		program        string
-		wantObjs       []mapstr.M
-		wantStatus     int
-		wantErr        bool
-		wantRawMessage []json.RawMessage
+		name       string
+		body       string
+		program    string
+		wantObjs   []mapstr.M
+		wantStatus int
+		wantErr    bool
 	}{
 		{
 			name:       "single object",
@@ -82,10 +80,6 @@ func Test_httpReadJSON(t *testing.T) {
 			name: "sequence of objects accepted (LF)",
 			body: `{"a":"1"}
 									{"a":"2"}`,
-			wantRawMessage: []json.RawMessage{
-				[]byte(`{"a":"1"}`),
-				[]byte(`{"a":"2"}`),
-			},
 			wantObjs:   []mapstr.M{{"a": "1"}, {"a": "2"}},
 			wantStatus: http.StatusOK,
 		},
@@ -110,26 +104,14 @@ func Test_httpReadJSON(t *testing.T) {
 			wantErr:    true,
 		},
 		{
-			name: "array of objects in stream",
-			body: `{"a":"1"} [{"a":"2"},{"a":"3"}] {"a":"4"}`,
-			wantRawMessage: []json.RawMessage{
-				[]byte(`{"a":"1"}`),
-				[]byte(`{"a":"2"}`),
-				[]byte(`{"a":"3"}`),
-				[]byte(`{"a":"4"}`),
-			},
+			name:       "array of objects in stream",
+			body:       `{"a":"1"} [{"a":"2"},{"a":"3"}] {"a":"4"}`,
 			wantObjs:   []mapstr.M{{"a": "1"}, {"a": "2"}, {"a": "3"}, {"a": "4"}},
 			wantStatus: http.StatusOK,
 		},
 		{
 			name: "numbers",
 			body: `{"a":1} [{"a":false},{"a":3.14}] {"a":-4}`,
-			wantRawMessage: []json.RawMessage{
-				[]byte(`{"a":1}`),
-				[]byte(`{"a":false}`),
-				[]byte(`{"a":3.14}`),
-				[]byte(`{"a":-4}`),
-			},
 			wantObjs: []mapstr.M{
 				{"a": int64(1)},
 				{"a": false},
@@ -171,13 +153,6 @@ func Test_httpReadJSON(t *testing.T) {
 				"timestamp": string(obj.timestamp), // leave timestamp in unix milli for ingest to handle.
 				"event": r,
 			})`,
-			wantRawMessage: []json.RawMessage{
-				[]byte(`{"event":{"data":"aGVsbG8=","number":1},"requestId":"ed4acda5-034f-9f42-bba1-f29aea6d7d8f","timestamp":"1578090901599"}`),
-				[]byte(`{"event":{"data":"c21hbGwgd29ybGQ=","number":9007199254740991},"requestId":"ed4acda5-034f-9f42-bba1-f29aea6d7d8f","timestamp":"1578090901599"}`),
-				[]byte(`{"event":{"data":"aGVsbG8gd29ybGQ=","number":"9007199254740992"},"requestId":"ed4acda5-034f-9f42-bba1-f29aea6d7d8f","timestamp":"1578090901599"}`),
-				[]byte(`{"event":{"data":"YmlnIHdvcmxk","number":"9223372036854775808"},"requestId":"ed4acda5-034f-9f42-bba1-f29aea6d7d8f","timestamp":"1578090901599"}`),
-				[]byte(`{"event":{"data":"d2lsbCBpdCBiZSBmcmllbmRzIHdpdGggbWU=","number":3.14},"requestId":"ed4acda5-034f-9f42-bba1-f29aea6d7d8f","timestamp":"1578090901599"}`),
-			},
 			wantObjs: []mapstr.M{
 				{"event": map[string]any{"data": "aGVsbG8=", "number": int64(1)}, "requestId": "ed4acda5-034f-9f42-bba1-f29aea6d7d8f", "timestamp": "1578090901599"},
 				{"event": map[string]any{"data": "c21hbGwgd29ybGQ=", "number": int64(9007199254740991)}, "requestId": "ed4acda5-034f-9f42-bba1-f29aea6d7d8f", "timestamp": "1578090901599"},
@@ -194,7 +169,7 @@ func Test_httpReadJSON(t *testing.T) {
 			if err != nil {
 				t.Fatalf("failed to compile program: %v", err)
 			}
-			gotObjs, rawMessages, gotStatus, err := httpReadJSON(strings.NewReader(tt.body), prg)
+			gotObjs, gotStatus, err := httpReadJSON(strings.NewReader(tt.body), prg)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("httpReadJSON() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -205,10 +180,6 @@ func Test_httpReadJSON(t *testing.T) {
 			if gotStatus != tt.wantStatus {
 				t.Errorf("httpReadJSON() gotStatus = %v, want %v", gotStatus, tt.wantStatus)
 			}
-			if tt.wantRawMessage != nil {
-				assert.Equal(t, tt.wantRawMessage, rawMessages)
-			}
-			assert.Equal(t, len(gotObjs), len(rawMessages))
 		})
 	}
 }