[receiver/awsfirehose] Unmarshal error when request is larger than the buffer limit of bufio.Scanner

[yiquanzhou]

Component(s)

No response

What happened?

Description

We are seeing error Error scanning for newline-delimited JSON with root cause bufio.Scanner: token too long in awsfirehosereceiver.

The unmarshaller uses

scanner := bufio.NewScanner(r)

to read and deserialize json payload of Firehose stream HTTP request. The scanner has an internal buffer which has the max size of 64KB.

However, Firehose buffers incoming data (CloudWatch logs & metrics) before sending them. Default buffer size is 5MB but configurable up to 64MB.

Each log event can take up to 256KB. doc

So we can hit the 64KB buffer quickly even with the default Firehose buffer size of 5MB.

Steps to Reproduce

Can be reproduced with a unit test which creates a message that is larger than 64KB

func TestUnmarshalLargePayload(t *testing.T) {
	unmarshaler := NewUnmarshaler(zap.NewNop(), component.NewDefaultBuildInfo())

	var largePayload bytes.Buffer
	largePayload.WriteString(`{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test","logStream":"test","logEvents":[`)
	largePayload.WriteString(`{"timestamp":1742239784,"message":"`)
	for largePayload.Len() < 64*1024-4 {
		largePayload.WriteString("a")
	}
	largePayload.WriteString(`"}]}`)

	compressedRecord, err := gzipData(largePayload.Bytes())
	require.NoError(t, err)

	_, err = unmarshaler.UnmarshalLogs(compressedRecord)
	require.Error(t, err)
	require.Equal(t, errInvalidRecords, err)
}

Expected Result

awsfirehose receiver should be able to handle large incoming requests without error.

How to fix?

Option1: raise buffer limit of bufio.Scanner

We can call https://pkg.go.dev/bufio#Scanner.Buffer method to pass a byte buffer with higher capacity to the scanner.

scanner := bufio.NewScanner(r)
buf := make([]byte, 0, initialBufferCapacity)
scanner.Buffer(buf, 64*1024*1024) // set the buffer limit to 64MB

We would need to determine the initial buffer capacity which ideally work for most of the requests. But this really depends on the data that is streamed by firehose. We might need to make this configurable.

Option2: use bufio.Reader

bufio.Reader seems to provide better error handling and can read large tokens. https://chriswilcox.dev/blog/2024/04/09/Scan-vs-Read-in-bufio.html

Collector version

v0.121.0

Environment information

Environment

OS: (e.g., "Macos 15.3.1")

OpenTelemetry Collector configuration

Log output

Additional context

No response

[github-actions]

Pinging code owners:

• receiver/awsfirehose: @Aneurysm9 @axw

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[axw]

Thanks for the report, and sorry for the bug. Option 2 sounds OK to me.

Perhaps we don't need to explicitly split by line though? encoding/json.Decoder.Decode will decode the next JSON object, so you can decode newline-delimited JSON by calling it in a loop. I don't know off hand whether json-iterator supports that too. I'll take a look at it tomorrow, unless you would like to take a stab at fixing it.

[yiquanzhou]

Perhaps we don't need to explicitly split by line though?

I was wondering if the request payload has multiple lines. The current implementation supports reading multiple json records such as

opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/testdata/multiple_records

Lines 1 to 2 in 9dbe5e0

	{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test","logStream":"test","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695232","timestamp":1725544035523,"message":"Hello world, here is our first log message!"}]}
	{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test","logStream":"test","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695233","timestamp":1725554035523,"message":"Hello world, here is our second log message!"}]}

But I configured CloudWatch to send logs to firehose and the request dump shows that the request payload is a json payload of schema https://docs.aws.amazon.com/firehose/latest/dev/httpdeliveryrequestresponse.html

title: FirehoseCustomHttpsEndpointRequest
description: >
  The request body that the Firehose service sends to
  custom HTTPS endpoints.
type: object
properties:
  requestId:
    description: >
      Same as the value in the X-Amz-Firehose-Request-Id header,
      duplicated here for convenience.
    type: string
  timestamp:
    description: >
      The timestamp (milliseconds since epoch) at which the Firehose
      server generated this request.
    type: integer
  records:
    description: >
      The actual records of the Firehose stream, carrying 
      the customer data.
    type: array
    minItems: 1
    maxItems: 10000
    items:
      type: object
      properties:
        data:
          description: >
            The data of this record, in Base64. Note that empty
            records are permitted in Firehose. The maximum allowed
            size of the data, before Base64 encoding, is 1024000
            bytes; the maximum length of this field is therefore
            1365336 chars.
          type: string
          minLength: 0
          maxLength: 1365336

required:
  - requestId
  - records

We can have multiple records in each request payload and each record has a data field. The data field's value is gzip encoded string. If we decode the data string, we get something like

{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test","logStream":"test","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695232","timestamp":1725544035523,"message":"Hello world, here is our first log message!"}]}

which can contain multiple logEvents.

My understanding is that the Unmarshaller only needs to read the json object above which has the logEvents field but this should fit in a single line.

opentelemetry-collector-contrib/receiver/awsfirehosereceiver/receiver.go

Line 223 in 81c5eee

statusCode, err := fmr.consumer.Consume(ctx, nextRecord, commonAttributes)

[axw]

@yiquanzhou yes, good catch! Now that you mention it, only CloudWatch Metric Streams produces newline-delimited JSON.

[yiquanzhou]

To support reading newline-delimited JSON, could we use https://pkg.go.dev/bufio#Reader.ReadBytes and pass \n as the delimiter ?

[axw]

@yiquanzhou that's certainly an option, but I don't think it's necessary for metric streams - the lines should be relatively short there. For logs we can just remove the scanning as you pointed out.

Feel free to send a PR and I'll review, otherwise I'll try to get a PR out in the next day or so.