x-pack/filebeat/inputs/httpjson: make response body decoding errors m…

…ore informative

The default rendering of errors from the json, csv and xml decoders can
be a little terse, but the error values themselves contain information
that allows the text context of the error to be constructed and
returned. It is a common problem that users are unable to decipher the
error messages, so add this text context to help.

CHANGELOG.next.asciidoc

@@ -208,6 +208,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - [Azure] Add input metrics to the azure-eventhub input. {pull}35739[35739]
 - Reduce HTTPJSON metrics allocations. {pull}36282[36282]
 - Add support for a simplified input configuraton when running under Elastic-Agent {pull}36390[36390]
+- Make HTTPJSON response body decoding errors more informative. {pull}[]
 
 *Auditbeat*
 

x-pack/filebeat/input/httpjson/encoding.go

@@ -9,7 +9,9 @@ import (
 	"bytes"
 	"encoding/csv"
 	"encoding/json"
+	stdxml "encoding/xml"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 
@@ -72,7 +74,11 @@ func encodeAsJSON(trReq transformable) ([]byte, error) {
 
 // decodeAsJSON decodes the JSON message in p into dst.
 func decodeAsJSON(p []byte, dst *response) error {
-	return json.Unmarshal(p, &dst.body)
+	err := json.Unmarshal(p, &dst.body)
+	if err != nil {
+		return jsonError{error: err, body: p}
+	}
+	return nil
 }
 
 // encodeAsForm encodes trReq as a URL encoded form.
@@ -95,14 +101,36 @@ func decodeAsNdjson(p []byte, dst *response) error {
 	for dec.More() {
 		var o interface{}
 		if err := dec.Decode(&o); err != nil {
-			return err
+			return jsonError{error: err, body: p}
 		}
 		results = append(results, o)
 	}
 	dst.body = results
 	return nil
 }
 
+type jsonError struct {
+	error
+	body []byte
+}
+
+func (e jsonError) Error() string {
+	switch err := e.error.(type) {
+	case nil:
+		return "<nil>"
+	case *json.SyntaxError:
+		return fmt.Sprintf("%v: text context %q", err, textContext(e.body, err.Offset))
+	case *json.UnmarshalTypeError:
+		return fmt.Sprintf("%v: text context %q", err, textContext(e.body, err.Offset))
+	default:
+		return err.Error()
+	}
+}
+
+func (e jsonError) Unwrap() error {
+	return e.error
+}
+
 // decodeAsCSV decodes p as a headed CSV document into dst.
 func decodeAsCSV(p []byte, dst *response) error {
 	var results []interface{}
@@ -135,7 +163,7 @@ func decodeAsCSV(p []byte, dst *response) error {
 
 	if err != nil {
 		if err != io.EOF { //nolint:errorlint // csv.Reader never wraps io.EOF.
-			return err
+			return csvError{error: err, body: p}
 		}
 	}
 
@@ -144,6 +172,31 @@ func decodeAsCSV(p []byte, dst *response) error {
 	return nil
 }
 
+type csvError struct {
+	error
+	body []byte
+}
+
+func (e csvError) Error() string {
+	switch err := e.error.(type) {
+	case nil:
+		return "<nil>"
+	case *csv.ParseError:
+		lines := bytes.Split(e.body, []byte{'\n'})
+		l := err.Line - 1 // Lines are 1-based.
+		if uint(l) >= uint(len(lines)) {
+			return err.Error()
+		}
+		return fmt.Sprintf("%v: text context %q", err, textContext(lines[l], int64(err.Column)))
+	default:
+		return err.Error()
+	}
+}
+
+func (e csvError) Unwrap() error {
+	return e.error
+}
+
 // decodeAsZip decodes p as a ZIP archive into dst.
 func decodeAsZip(p []byte, dst *response) error {
 	var results []interface{}
@@ -165,7 +218,7 @@ func decodeAsZip(p []byte, dst *response) error {
 			var o interface{}
 			if err := dec.Decode(&o); err != nil {
 				rc.Close()
-				return err
+				return jsonError{error: err, body: p}
 			}
 			results = append(results, o)
 		}
@@ -185,9 +238,73 @@ func decodeAsZip(p []byte, dst *response) error {
 func decodeAsXML(p []byte, dst *response) error {
 	cdata, body, err := xml.Unmarshal(bytes.NewReader(p), dst.xmlDetails)
 	if err != nil {
-		return err
+		return xmlError{error: err, body: p}
 	}
 	dst.body = body
 	dst.header["XML-CDATA"] = []string{cdata}
 	return nil
 }
+
+type xmlError struct {
+	error
+	body []byte
+}
+
+func (e xmlError) Error() string {
+	switch err := e.error.(type) {
+	case nil:
+		return "<nil>"
+	case *stdxml.SyntaxError:
+		lines := bytes.Split(e.body, []byte{'\n'})
+		l := err.Line - 1 // Lines are 1-based.
+		if uint(l) >= uint(len(lines)) {
+			return err.Error()
+		}
+		// The xml package does not provide column-level context,
+		// so just point to the start of the line.
+		return fmt.Sprintf("%v: text context %q", err, textContext(lines[l], 0))
+	default:
+		return err.Error()
+	}
+}
+
+func (e xmlError) Unwrap() error {
+	return e.error
+}
+
+// textContext returns the context of text around the provided position starting
+// five bytes before pos and extending ten bytes, dependent on the length of the
+// text and the value of pos relative to bounds. If a text truncation is made,
+// an ellipsis is added to indicate this. The returned []byte should not be mutated
+// as it may be shared with the caller.
+func textContext(text []byte, pos int64) []byte {
+	left := maxInt64(0, pos-5)
+	text = text[left:]
+	var pad int64
+	if left != 0 {
+		pad = 3
+		text = append([]byte("..."), text...)
+	}
+	right := minInt(pos+10+pad, int64(len(text)))
+	if right != int64(len(text)) {
+		// Ensure we don't clobber the body's bytes.
+		text = append(text[:right:right], []byte("...")...)
+	} else {
+		text = text[:right]
+	}
+	return text
+}
+
+func minInt(a, b int64) int64 {
+	if a < b {
+		return a
+	}
+	return b
+}
+
+func maxInt64(a, b int64) int64 {
+	if a > b {
+		return a
+	}
+	return b
+}