Make StreamingLogsUnmarshaler optional

Author: MichaelKatsoulis

extension/encoding/awslogsencodingextension/extension.go

@@ -255,5 +255,8 @@ func (e *encodingExtension) UnmarshalLogs(buf []byte) (plog.Logs, error) {
 }
 
 func (e *encodingExtension) NewLogsDecoder(reader io.Reader, options ...encoding.DecoderOption) (encoding.LogsDecoder, error) {
-	return e.unmarshaler.NewLogsDecoder(reader, options...)
+	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)
 }

extension/encoding/awslogsencodingextension/internal/unmarshaler/cloudtraillog/unmarshaler.go

@@ -33,7 +33,10 @@ type CloudTrailLogUnmarshaler struct {
 	uIDFeatureEnabled bool
 }
 
-var _ unmarshaler.AWSUnmarshaler = (*CloudTrailLogUnmarshaler)(nil)
+var (
+	_ unmarshaler.AWSUnmarshaler           = (*CloudTrailLogUnmarshaler)(nil)
+	_ unmarshaler.StreamingLogsUnmarshaler = (*CloudTrailLogUnmarshaler)(nil)
+)
 
 // UserIdentity represents the user identity information in CloudTrail logs
 type UserIdentity struct {

extension/encoding/awslogsencodingextension/internal/unmarshaler/unmarshaler.go

@@ -11,7 +11,13 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding"
 )
 
+// AWSUnmarshaler is the base interface for all AWS log format unmarshalers.
 type AWSUnmarshaler interface {
 	UnmarshalAWSLogs(reader io.Reader) (plog.Logs, error)
+}
+
+// StreamingLogsUnmarshaler is optionally implemented by unmarshalers that support streaming decoding.
+type StreamingLogsUnmarshaler interface {
+	AWSUnmarshaler
 	NewLogsDecoder(reader io.Reader, options ...encoding.DecoderOption) (encoding.LogsDecoder, error)
 }