Implement streaming in cloudtrail format of awslogsencoding extension#46203
Implement streaming in cloudtrail format of awslogsencoding extension#46203MichaelKatsoulis wants to merge 10 commits intoopen-telemetry:mainfrom
Conversation
…soulis/opentelemetry-collector-contrib into feat/streaming-for-cloudtrail
| //nolint:errorlint | ||
| if err == io.EOF { |
There was a problem hiding this comment.
Why do you supress this lint error?
Can you replace it with error.Is
There was a problem hiding this comment.
Good catch! Done
There was a problem hiding this comment.
This was there to avoid checking wrapped errors. Here the error get ignored if and only if it's an unwrapped io.EOF. And the expectation is to get this from the streaming implementation for EOF.
And we can get EOF wrapped from the stream itself. We should not ignore those errors and consider them as failures. This is the reason logic to be like this
@constanca-m @MichaelKatsoulis so we need this unwrapped EOF check
There was a problem hiding this comment.
Thank you. Then can you also add a comment on the code about it? So we don't take it out again
There was a problem hiding this comment.
@constanca-m yup, makes sense to add a comment 💯
constanca-m
left a comment
constanca-m
left a comment
There was a problem hiding this comment.
Looks good, couldn't find any issues. @Kavindu-Dodan please review as well
|
@Kavindu-Dodan look into this commit a4350e1 I had to make streaming optional for the |
| func (e *encodingExtension) NewLogsDecoder(reader io.Reader, options ...encoding.DecoderOption) (encoding.LogsDecoder, error) { | ||
| if u, ok := e.unmarshaler.(awsunmarshaler.StreamingLogsUnmarshaler); ok { | ||
| return u.NewLogsDecoder(reader, options...) | ||
| } | ||
| return nil, fmt.Errorf("streaming not supported for format %q", e.format) | ||
| } |
There was a problem hiding this comment.
@constanca-m @MichaelKatsoulis Should even add a signal to this PR ? IMO we can further split it to have API contract in this PR and isolate all signals to individual PRs :)
There was a problem hiding this comment.
See #46211, this allows us to build all signal related PRs based on that specific PR. IMO it's easier that way
axw
left a comment
axw
left a comment
There was a problem hiding this comment.
Needs an update regarding the offset. I suggest we continue in @Kavindu-Dodan's PR, and close this one.
| isEOF = true | ||
| return record, nil | ||
| } | ||
| return xstreamencoding.NewLogsDecoderAdapter(decoderF, func() int64 { return 0 }), nil |
There was a problem hiding this comment.
Returning 0 is invalid per the contract:
// Offset returns the offset after the most recent batch read from the stream, or the initial offset.
... though the wording probably needs changing to clarify that "or the initial offset" is only in the case where no data has been flushed yet.
Should we continue this in #46212?
|
Closing this and continue the implementation of adoption in PRs #46203 (comment) |
5 participants
Description
Adopting encoder extension stream decoding added through #45567
This PR focuses only on CloudTrail log Format of
awslogsencodingextension