internal/testrunner/runners/pipeline: unmarshal test results using UseNumber

internal/testrunner/runners/pipeline/runner.go

@@ -103,6 +103,8 @@ func (r *runner) run() ([]testrunner.TestResult, error) {
 		}
 		startTime := time.Now()
 
+		// TODO: Add tests to cover regressive use of json.Unmarshal in loadTestCaseFile.
+		// See https://github.com/elastic/elastic-package/pull/717.
 		tc, err := r.loadTestCaseFile(testCaseFile)
 		if err != nil {
 			err := errors.Wrap(err, "loading test case failed")
@@ -141,6 +143,8 @@ func (r *runner) run() ([]testrunner.TestResult, error) {
 			return nil, errors.Wrapf(err, "creating fields validator for data stream failed (path: %s, test case file: %s)", dataStreamPath, testCaseFile)
 		}
 
+		// TODO: Add tests to cover regressive use of json.Unmarshal in verifyResults.
+		// See https://github.com/elastic/elastic-package/pull/717.
 		err = r.verifyResults(testCaseFile, tc.config, result, fieldsValidator)
 		if e, ok := err.(testrunner.ErrTestCaseFailed); ok {
 			tr.FailureMsg = e.Error()
@@ -226,6 +230,8 @@ func (r *runner) verifyResults(testCaseFile string, config *testConfig, result *
 	testCasePath := filepath.Join(r.options.TestFolder.Path, testCaseFile)
 
 	if r.options.GenerateTestResult {
+		// TODO: Add tests to cover regressive use of json.Unmarshal in writeTestResult.
+		// See https://github.com/elastic/elastic-package/pull/717.
 		err := writeTestResult(testCasePath, result)
 		if err != nil {
 			return errors.Wrap(err, "writing test result failed")
@@ -275,7 +281,7 @@ func verifyDynamicFields(result *testResult, config *testConfig) error {
 	var multiErr multierror.Error
 	for _, event := range result.events {
 		var m common.MapStr
-		err := json.Unmarshal(event, &m)
+		err := jsonUnmarshalUsingNumber(event, &m)
 		if err != nil {
 			return errors.Wrap(err, "can't unmarshal event")
 		}
@@ -342,7 +348,7 @@ func checkErrorMessage(event json.RawMessage) error {
 			Message interface{}
 		}
 	}
-	err := json.Unmarshal(event, &pipelineError)
+	err := jsonUnmarshalUsingNumber(event, &pipelineError)
 	if err != nil {
 		return errors.Wrapf(err, "can't unmarshal event to check pipeline error: %#q", event)
 	}

internal/testrunner/runners/pipeline/runner_test.go

@@ -6,6 +6,8 @@ package pipeline
 
 import (
 	"encoding/json"
+	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -318,3 +320,108 @@ func TestDiffUlite(t *testing.T) {
 		})
 	}
 }
+
+var jsonUnmarshalUsingNumberTests = []struct {
+	name string
+	msg  string
+}{
+	{
+		name: "empty",
+		msg:  "", // Will error "unexpected end of JSON input".
+	},
+	{
+		name: "string",
+		msg:  `"message"`,
+	},
+	{
+		name: "array",
+		msg:  "[1,2,3,4,5]",
+	},
+	{
+		name: "object",
+		msg:  `{"key":42}`,
+	},
+	{
+		name: "object",
+		msg:  `{"key":42}answer`, // Will error "invalid character 'a' after top-level value".
+	},
+	// Test extra data whitespace parity with json.Unmarshal for error parity.
+	{
+		name: "object",
+		msg:  `{"key":42} `,
+	},
+	{
+		name: "object",
+		msg:  `{"key":42}` + "\t",
+	},
+	{
+		name: "object",
+		msg:  `{"key":42}` + "\r",
+	},
+	{
+		name: "object",
+		msg:  `{"key":42}` + "\n",
+	},
+	{
+		name: "0x1p52+1",
+		msg:  fmt.Sprint(uint64(0x1p52) + 1),
+	},
+	{
+		name: "0x1p53-1",
+		msg:  fmt.Sprint(uint64(0x1p53) - 1),
+	},
+	// The following three cases will fail if json.Unmarshal is used in place
+	// of jsonUnmarshalUsingNumber, as they are past the cutover.
+	{
+		name: "0x1p53+1",
+		msg:  fmt.Sprint(uint64(0x1p53) + 1),
+	},
+	{
+		name: "0x1p54+1",
+		msg:  fmt.Sprint(uint64(0x1p54) + 1),
+	},
+	{
+		name: "long",
+		msg:  "9223372036854773807",
+	},
+}
+
+func TestJsonUnmarshalUsingNumberRoundTrip(t *testing.T) {
+	// This tests that jsonUnmarshalUsingNumber behaves the same
+	// way as json.Unmarshal with the exception that numbers are
+	// not unmarshaled through float64. This is important to avoid
+	// low-bit truncation of long numeric values that are greater
+	// than or equal to 0x1p53, the limit of bijective equivalence
+	// with 64 bit-integers.
+
+	for _, test := range jsonUnmarshalUsingNumberTests {
+		t.Run(test.name, func(t *testing.T) {
+			var val interface{}
+			err := jsonUnmarshalUsingNumber([]byte(test.msg), &val)
+
+			// Confirm that we get the same errors with jsonUnmarshalUsingNumber
+			// as are returned by json.Unmarshal.
+			jerr := json.Unmarshal([]byte(test.msg), new(interface{}))
+			if (err == nil) != (jerr == nil) {
+				t.Errorf("unexpected error: got:%#v want:%#v", err, jerr)
+			}
+			if err != nil {
+				return
+			}
+
+			// Confirm that we round-trip the message correctly without
+			// alteration beyond trailing whitespace.
+			got, err := json.Marshal(val)
+			if err != nil {
+				t.Errorf("unexpected error: got:%#v want:%#v", err, jerr)
+			}
+			// Truncate trailing whitespace from the input since it won't
+			// be rendered in the output. This set of space characters is
+			// defined in encoding/json/scanner.go as func isSpace.
+			want := strings.TrimRight(test.msg, " \t\r\n")
+			if string(got) != want {
+				t.Errorf("unexpected result: got:%v want:%v", val, want)
+			}
+		})
+	}
+}

internal/testrunner/runners/pipeline/test_case.go

@@ -28,7 +28,7 @@ type testCaseDefinition struct {
 
 func readTestCaseEntriesForEvents(inputData []byte) ([]json.RawMessage, error) {
 	var tcd testCaseDefinition
-	err := json.Unmarshal(inputData, &tcd)
+	err := jsonUnmarshalUsingNumber(inputData, &tcd)
 	if err != nil {
 		return nil, errors.Wrap(err, "unmarshalling input data failed")
 	}
@@ -59,7 +59,7 @@ func createTestCase(filename string, entries []json.RawMessage, config *testConf
 	var events []json.RawMessage
 	for _, entry := range entries {
 		var m common.MapStr
-		err := json.Unmarshal(entry, &m)
+		err := jsonUnmarshalUsingNumber(entry, &m)
 		if err != nil {
 			return nil, errors.Wrap(err, "can't unmarshal test case entry")
 		}

internal/testrunner/runners/pipeline/test_result.go

@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -188,7 +189,7 @@ func adjustTestResult(result *testResult, config *testConfig) (*testResult, erro
 		}
 
 		var m common.MapStr
-		err := json.Unmarshal(event, &m)
+		err := jsonUnmarshalUsingNumber(event, &m)
 		if err != nil {
 			return nil, errors.Wrapf(err, "can't unmarshal event: %s", string(event))
 		}
@@ -212,7 +213,7 @@ func adjustTestResult(result *testResult, config *testConfig) (*testResult, erro
 
 func unmarshalTestResult(body []byte) (*testResult, error) {
 	var trd testResultDefinition
-	err := json.Unmarshal(body, &trd)
+	err := jsonUnmarshalUsingNumber(body, &trd)
 	if err != nil {
 		return nil, errors.Wrap(err, "unmarshalling test result failed")
 	}
@@ -222,6 +223,28 @@ func unmarshalTestResult(body []byte) (*testResult, error) {
 	return &tr, nil
 }
 
+// jsonUnmarshalUsingNumber is a drop-in replacement for json.Unmarshal that
+// does not default to unmarshaling numeric values to float64 in order to
+// prevent low bit truncation of values greater than 1<<53.
+// See https://golang.org/cl/6202068 for details.
+func jsonUnmarshalUsingNumber(data []byte, v interface{}) error {
+	dec := json.NewDecoder(bytes.NewReader(data))
+	dec.UseNumber()
+	err := dec.Decode(v)
+	if err != nil {
+		if err == io.EOF {
+			return errors.New("unexpected end of JSON input")
+		}
+		return err
+	}
+	// Make sure there is no more data after the message
+	// to approximate json.Unmarshal's behaviour.
+	if dec.More() {
+		return fmt.Errorf("more data after top-level value")
+	}
+	return nil
+}
+
 func marshalTestResultDefinition(result *testResult) ([]byte, error) {
 	var trd testResultDefinition
 	trd.Expected = result.events
@@ -241,7 +264,7 @@ func marshalNormalizedJSON(v testResultDefinition) ([]byte, error) {
 		return msg, err
 	}
 	var obj interface{}
-	err = json.Unmarshal(msg, &obj)
+	err = jsonUnmarshalUsingNumber(msg, &obj)
 	if err != nil {
 		return msg, err
 	}

test/packages/other/long_integers/_dev/build/build.yml

@@ -0,0 +1,3 @@
+dependencies:
+  ecs:
+    reference: [email protected]

test/packages/other/long_integers/_dev/build/docs/README.md

@@ -0,0 +1,5 @@
+# Long Integer Tests
+
+{{event "test"}}
+
+{{fields "test"}}

test/packages/other/long_integers/changelog.yml

@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "0.0.1"
+  changes:
+    - description: Initial draft of the package
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/0 # FIXME Replace with the real PR link

test/packages/other/long_integers/data_stream/test/_dev/test/pipeline/test-sequences.log

@@ -0,0 +1 @@
+4503599627370497,9007199254740991,9007199254740993,18014398509481985,9223372036854773807

test/packages/other/long_integers/data_stream/test/_dev/test/pipeline/test-sequences.log-config.yml

@@ -0,0 +1,2 @@
+fields:
+  "@warning": "The values in sequence_number must match the values in message."

test/packages/other/long_integers/data_stream/test/_dev/test/pipeline/test-sequences.log-expected.json

@@ -0,0 +1,15 @@
+{
+    "expected": [
+        {
+            "@warning": "The values in sequence_number must match the values in message.",
+            "message": "4503599627370497,9007199254740991,9007199254740993,18014398509481985,9223372036854773807",
+            "sequence_number": [
+                4503599627370497,
+                9007199254740991,
+                9007199254740993,
+                18014398509481985,
+                9223372036854773807
+            ]
+        }
+    ]
+}

test/packages/other/long_integers/data_stream/test/agent/stream/stream.yml.hbs

@@ -0,0 +1,7 @@
+paths:
+{{#each paths as |path i|}}
+  - {{path}}
+{{/each}}
+exclude_files: [".gz$"]
+processors:
+  - add_locale: ~

test/packages/other/long_integers/data_stream/test/elasticsearch/ingest_pipeline/default.yml

@@ -0,0 +1,16 @@
+---
+description: Pipeline for processing sample logs
+processors:
+- split:
+    field: message
+    separator: ","
+    target_field: sequence_number
+    ignore_missing: true
+- convert:
+    field: sequence_number
+    type: long
+
+on_failure:
+- set:
+    field: error.message
+    value: '{{ _ingest.on_failure_message }}'

test/packages/other/long_integers/data_stream/test/fields/base-fields.yml

@@ -0,0 +1,22 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: Event timestamp.
+- name: '@warning'
+  type: keyword
+  description: Warning for devs.
+- name: message
+  type: keyword
+  description: Original input.
+- name: sequence_number
+  type: long
+  description: |
+    Log entry identifier that is incremented sequentially. Unique for each log type.

test/packages/other/long_integers/data_stream/test/manifest.yml

@@ -0,0 +1,13 @@
+title: "Test"
+type: logs
+streams:
+  - input: logfile
+    title: Sample logs
+    description: Collect sample logs
+    vars:
+      - name: paths
+        type: text
+        title: Paths
+        multi: true
+        default:
+          - /var/log/*.log

test/packages/other/long_integers/data_stream/test/sample_event.json

@@ -0,0 +1,11 @@
+{
+    "@warning": "The values in sequence_number must match the values in message.",
+    "message": "4503599627370497,9007199254740991,9007199254740993,18014398509481985,9223372036854773807",
+    "sequence_number": [
+        4503599627370497,
+        9007199254740991,
+        9007199254740993,
+        18014398509481985,
+        9223372036854773807
+    ]
+}

test/packages/other/long_integers/docs/README.md

@@ -0,0 +1,29 @@
+# Long Integer Tests
+
+An example event for `test` looks as following:
+
+```json
+{
+    "@warning": "The values in sequence_number must match the values in message.",
+    "message": "4503599627370497,9007199254740991,9007199254740993,18014398509481985,9223372036854773807",
+    "sequence_number": [
+        4503599627370497,
+        9007199254740991,
+        9007199254740993,
+        18014398509481985,
+        9223372036854773807
+    ]
+}
+```
+
+**Exported fields**
+
+| Field | Description | Type |
+|---|---|---|
+| @timestamp | Event timestamp. | date |
+| @warning | Warning for devs. | keyword |
+| data_stream.dataset | Data stream dataset. | constant_keyword |
+| data_stream.namespace | Data stream namespace. | constant_keyword |
+| data_stream.type | Data stream type. | constant_keyword |
+| message | Original input. | keyword |
+| sequence_number | Log entry identifier that is incremented sequentially. Unique for each log type. | long |