feat: enhance receiver & handler behaviour

Signed-off-by: Kavindu Dodanduwa <[email protected]>

Author: Kavindu-Dodan

receiver/awslambdareceiver/README.md

@@ -54,44 +54,19 @@ The `awslambdareceiver` operates as follows:
 
 The following receiver configuration parameters are supported.
 
-| Name          | Description                                                                                                                                                                                        | Required |
-|:--------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|
-| `s3_encoding` | Name of the encoding extension to use for S3 objects. For example, `awslogs_encoding` to use AWS logs encoding extensions. When unspecified, falls back to CloudWatch subscription filter handling | Optional |
+| Name                 | Description                                                                                                        |
+|:---------------------|:-------------------------------------------------------------------------------------------------------------------|
+| `encoding_extension` | Required: The name of the encoding extension to load which will decode the data received through supported sources | 
 
 ### Example Configuration
 
-```yaml
-receivers:
-  awslambda:
-    s3_encoding: awslogs_encoding
-
-extensions:
-  awslogs_encoding:
-    format: vpcflow
-    vpcflow:
-      file_format: plain-text
-
-exporters:
-  otlphttp:
-    endpoint: "https://my-backend:443"
-
-service:
-  extensions:
-    - awslogs_encoding
-  pipelines:
-    logs:
-      receivers: [awslambda]
-      exporters: [otlphttp]
-```
-
-## Examples
 
 ### Example 1: VPC Flow Logs from S3
 
 ```yaml
 receivers:
   awslambda:
-    s3_encoding: awslogs_encoding
+    encoding_extension: awslogs_encoding
 
 extensions:
   awslogs_encoding:
@@ -112,14 +87,16 @@ service:
       exporters: [otlphttp]
 ```
 
-In this example, `awslambdareceiver` receives a notification when a new VPC flow log file is stored in an S3 bucket. The receiver fetches the log file from S3 and parses it using the `awslogs_encoding` extension with vpcflow format. The parsed logs are then sent to an OTLP listener using the `otlphttp` exporter.
+In this example, the `awslambdareceiver` is expected to be triggered when a VPC flow log is created at S3 bucket.
+The receiver retrieves the log file from S3 and decodes it using the `awslogs_encoding` extension with the vpcflow format.
+Parsed logs are forwarded to an OTLP listener via the `otlphttp` exporter.
 
 ### Example 2: ELB Access Logs from S3
 
 ```yaml
 receivers:
   awslambda:
-    s3_encoding: awslogs_encoding
+    encoding_extension: awslogs_encoding
 
 extensions:
   awslogs_encoding:
@@ -140,32 +117,37 @@ service:
       exporters: [otlphttp]
 ```
 
-### Example 3: CloudWatch Logs Subscription
+### Example 3: CloudWatch Logs mode
 
 ```yaml
 receivers:
   awslambda:
+    encoding_extension: awslogs_encoding
+
+extensions:
+  awslogs_encoding:
+    format: cloudwatch
 
 exporters:
   otlphttp:
     endpoint: "https://my-backend:443"
 
 service:
+  extensions:
+    - awslogs_encoding
   pipelines:
     logs:
       receivers: [awslambda]
       exporters: [otlphttp]
 ```
 
-In this example, `awslambdareceiver` is invoked by a CloudWatch Logs subscription filter. 
-`s3_encoding` configuration is omitted since it is not needed for CloudWatch Logs.
-The receiver automatically parses the CloudWatch Logs data using the default `awslogs_encoding` extension with cloudwatch format. 
-No explicit encoding configuration is needed. The parsed logs are then sent to an OTLP listener using the `otlphttp` exporter.
+In this example, `awslambdareceiver` is expected to be triggered by a CloudWatch Logs subscription filter.
+The receiver retrieves the logs from the event payload and decodes them using the `awslogs_encoding` extension with the cloudwatch format.
 
 ## Supported Data Types
 
-- **Logs** (Primary support)
-- **Metrics** (Future consideration)
+- **Logs**: Supported through S3 and CloudWatch Logs event sources
+- **Metrics**: Supported through S3
 
 ## AWS Permissions
 

receiver/awslambdareceiver/benchmark_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"encoding/json"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-lambda-go/events"
 	"github.com/stretchr/testify/require"
@@ -40,8 +41,8 @@ func BenchmarkHandleS3Notification(b *testing.B) {
 
 	consumer := noOpLogsConsumer{}
 	// Wrap the consumer to match the new s3EventConsumerFunc signature
-	logsConsumer := func(ctx context.Context, event events.S3EventRecord, logs plog.Logs) error {
-		setObservedTimestampForAllLogs(logs, event.EventTime)
+	logsConsumer := func(ctx context.Context, time time.Time, logs plog.Logs) error {
+		setObservedTimestampForAllLogs(logs, time)
 		return consumer.ConsumeLogs(ctx, logs)
 	}
 	handler := newS3Handler(service, zap.NewNop(), mockS3LogUnmarshaler{}.UnmarshalLogs, logsConsumer)

receiver/awslambdareceiver/config.go

@@ -4,22 +4,19 @@
 package awslambdareceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awslambdareceiver"
 
 import (
-	"go.opentelemetry.io/collector/component"
-)
+	"errors"
 
-const (
-	awsLogsEncoding = "awslogs_encoding"
-	s3ARNPrefix     = "arn:aws:s3:::"
+	"go.opentelemetry.io/collector/component"
 )
 
 type Config struct {
-	// S3Encoding identifies the encoding of the S3 objects that trigger the Lambda.
+	// EncodingExtension defines the encoding extension the receiver must use to decode receiving data
 	//
-	// If S3 data is in multiple formats (ex:- VPC flow logs, CloudTrail logs), you should deploy
+	// If receiving data is in different formats(ex:- VPC flow logs, CloudTrail logs), receiver must be deployed in
 	// separate Lambda functions with specific extension configurations.
 	//
-	// If unspecified, the receiver falls back to work with CloudWatch Log subscription encoding extension.
-	S3Encoding string `mapstructure:"s3_encoding"`
+	// This property is mandatory and must refer to a valid encoding extension configured in the collector configurations.
+	EncodingExtension string `mapstructure:"encoding_extension"`
 
 	_ struct{} // Prevent unkeyed literal initialization
 }
@@ -30,6 +27,10 @@ func createDefaultConfig() component.Config {
 	return &Config{}
 }
 
-func (*Config) Validate() error {
+func (c *Config) Validate() error {
+	if c.EncodingExtension == "" {
+		return errors.New("encoding_extension is mandatory, please use a valid encoding extension name configured in the collector configurations")
+	}
+
 	return nil
 }

receiver/awslambdareceiver/config_test.go

@@ -28,22 +28,23 @@ func TestLoadConfig(t *testing.T) {
 		expectError string
 	}{
 		{
-			id: component.NewIDWithName(metadata.Type, awsLogsEncoding),
+			id: component.NewIDWithName(metadata.Type, "awslogs_encoding"),
 			expected: &Config{
-				S3Encoding: awsLogsEncoding,
+				EncodingExtension: "awslogs_encoding",
 			},
 		},
 		{
 			id: component.NewIDWithName(metadata.Type, "json_log_encoding"),
 			expected: &Config{
-				S3Encoding: "json_log_encoding",
+				EncodingExtension: "json_log_encoding",
 			},
 		},
 		{
 			id: component.NewIDWithName(metadata.Type, "empty_encoding"),
 			expected: &Config{
-				S3Encoding: "",
+				EncodingExtension: "",
 			},
+			expectError: "encoding_extension is mandatory",
 		},
 	}
 

receiver/awslambdareceiver/factory_test.go

@@ -14,7 +14,7 @@ import (
 
 func TestCreateLogsReceiver(t *testing.T) {
 	cfg := createDefaultConfig()
-	require.Empty(t, cfg.(*Config).S3Encoding)
+	cfg.(*Config).EncodingExtension = "awslogs_encoding"
 
 	r, err := createLogsReceiver(t.Context(), receivertest.NewNopSettings(metadata.Type), cfg, consumertest.NewNop())
 	require.NoError(t, err)

receiver/awslambdareceiver/handler.go

@@ -9,6 +9,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/aws/aws-lambda-go/events"
@@ -23,9 +24,51 @@ import (
 
 type (
 	unmarshalFunc[T any]       func([]byte) (T, error)
-	s3EventConsumerFunc[T any] func(context.Context, events.S3EventRecord, T) error
+	s3EventConsumerFunc[T any] func(context.Context, time.Time, T) error
+	handlerRegistry            map[eventType]func() lambdaEventHandler
 )
 
+type handlerProvider interface {
+	getHandler(eventType eventType) (lambdaEventHandler, error)
+}
+
+// handlerProvider is responsible for providing event handlers based on event types.
+// It operates with a registry of handler factories and caches loadedHandlers for reuse.
+type handlerProviderImpl struct {
+	registry       handlerRegistry
+	loadedHandlers map[eventType]lambdaEventHandler
+	knownTypes     []string
+}
+
+func newHandlerProvider(registry handlerRegistry) handlerProvider {
+	var types []string
+	for t := range registry {
+		types = append(types, string(t))
+	}
+
+	return &handlerProviderImpl{
+		loadedHandlers: map[eventType]lambdaEventHandler{},
+		registry:       registry,
+		knownTypes:     types,
+	}
+}
+
+func (h *handlerProviderImpl) getHandler(eventType eventType) (lambdaEventHandler, error) {
+	if loaded, exists := h.loadedHandlers[eventType]; exists {
+		return loaded, nil
+	}
+
+	factory, exists := h.registry[eventType]
+	if !exists {
+		return nil, fmt.Errorf("no handler registered for event type %s, known types: '%s'", eventType, strings.Join(h.knownTypes, ","))
+	}
+
+	handler := factory()
+	h.loadedHandlers[eventType] = handler
+	return handler, nil
+}
+
+// lambdaEventHandler defines the contract for AWS Lambda event handlers
 type lambdaEventHandler interface {
 	handlerType() eventType
 	handle(ctx context.Context, event json.RawMessage) error
@@ -68,7 +111,7 @@ func (s *s3Handler[T]) handle(ctx context.Context, event json.RawMessage) error
 		zap.String("S3Bucket", parsedEvent.S3.Bucket.Arn),
 	)
 
-	// Skip processing zero length objects. This includes events from folder creation and empty object .
+	// Skip processing zero length objects. This includes events from folder creation and empty object.
 	if parsedEvent.S3.Object.Size == 0 {
 		s.logger.Info("Empty object, skipping download", zap.String("File", parsedEvent.S3.Object.Key))
 		return nil
@@ -84,7 +127,7 @@ func (s *s3Handler[T]) handle(ctx context.Context, event json.RawMessage) error
 		return fmt.Errorf("failed to unmarshal logs: %w", err)
 	}
 
-	if err := s.consumer(ctx, parsedEvent, data); err != nil {
+	if err := s.consumer(ctx, parsedEvent.EventTime, data); err != nil {
 		// If permanent, return as-is (don't retry)
 		if consumererror.IsPermanent(err) {
 			return err

receiver/awslambdareceiver/handler_test.go

@@ -16,8 +16,10 @@ import (
 
 	"github.com/aws/aws-lambda-go/events"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
+	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	conventions "go.opentelemetry.io/otel/semconv/v1.27.0"
@@ -114,8 +116,8 @@ func TestProcessLambdaEvent_S3Notification(t *testing.T) {
 			s3Service.EXPECT().ReadObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(test.mockContent, nil).AnyTimes()
 
 			// Wrap the consumer to match the new s3EventConsumerFunc signature
-			logsConsumer := func(ctx context.Context, event events.S3EventRecord, logs plog.Logs) error {
-				setObservedTimestampForAllLogs(logs, event.EventTime)
+			logsConsumer := func(ctx context.Context, time time.Time, logs plog.Logs) error {
+				setObservedTimestampForAllLogs(logs, time)
 				return test.eventConsumer.ConsumeLogs(ctx, logs)
 			}
 
@@ -207,8 +209,8 @@ func TestS3HandlerParseEvent(t *testing.T) {
 
 	var consumer noOpLogsConsumer
 	// Wrap the consumer to match the new s3EventConsumerFunc signature
-	logsConsumer := func(ctx context.Context, event events.S3EventRecord, logs plog.Logs) error {
-		setObservedTimestampForAllLogs(logs, event.EventTime)
+	logsConsumer := func(ctx context.Context, time time.Time, logs plog.Logs) error {
+		setObservedTimestampForAllLogs(logs, time)
 		return consumer.ConsumeLogs(ctx, logs)
 	}
 	handler := newS3Handler(s3Service, zap.NewNop(), mockS3LogUnmarshaler{}.UnmarshalLogs, logsConsumer)
@@ -254,8 +256,27 @@ func TestHandleCloudwatchLogEvent(t *testing.T) {
 		},
 	}
 
-	unmarshaler, err := loadSubFilterLogUnmarshaler(t.Context(), awslogsencodingextension.NewFactory())
-	require.NoError(t, err)
+	var extensionID component.ID
+	err := extensionID.UnmarshalText([]byte("awslogs_encoding"))
+	if err != nil {
+		t.Errorf("failed to unmarshal identifier: %v", err)
+		return
+	}
+
+	factory := awslogsencodingextension.NewFactory()
+	ext, err := factory.Create(t.Context(), extension.Settings{ID: extensionID}, &awslogsencodingextension.Config{
+		Format: "cloudwatch",
+	})
+	if err != nil {
+		t.Errorf("failed to create awslogs encoding extension: %v", err)
+		return
+	}
+
+	subscriptionFilterLogUnmarshaler, ok := ext.(plog.Unmarshaler)
+	if !ok {
+		t.Errorf("extension does not implement log.Unmarshaler")
+		return
+	}
 
 	for name, test := range tests {
 		t.Run(name, func(t *testing.T) {
@@ -266,11 +287,11 @@ func TestHandleCloudwatchLogEvent(t *testing.T) {
 			}
 
 			var lambdaEvent json.RawMessage
-			lambdaEvent, err = json.Marshal(cwEvent)
+			lambdaEvent, err := json.Marshal(cwEvent)
 			require.NoError(t, err)
 
-			handler := newCWLogsSubscriptionHandler(zap.NewNop(), unmarshaler.UnmarshalLogs, test.eventConsumer.ConsumeLogs)
-			err := handler.handle(t.Context(), lambdaEvent)
+			handler := newCWLogsSubscriptionHandler(zap.NewNop(), subscriptionFilterLogUnmarshaler.UnmarshalLogs, test.eventConsumer.ConsumeLogs)
+			err = handler.handle(t.Context(), lambdaEvent)
 			if test.expectedErr != "" {
 				require.ErrorContains(t, err, test.expectedErr)
 			} else {
@@ -366,7 +387,7 @@ func TestConsumerErrorHandling(t *testing.T) {
 			s3Service.EXPECT().ReadObject(gomock.Any(), gomock.Any(), gomock.Any()).Return([]byte(mockContent), nil).Times(1)
 
 			// Consumer that returns the test error
-			logsConsumer := func(_ context.Context, _ events.S3EventRecord, _ plog.Logs) error {
+			logsConsumer := func(_ context.Context, _ time.Time, _ plog.Logs) error {
 				return test.consumerErr
 			}
 
@@ -432,6 +453,14 @@ func (mockS3LogUnmarshaler) UnmarshalLogs(data []byte) (plog.Logs, error) {
 	return plog.Logs{}, errors.New("logs not in the correct format")
 }
 
+type mockHandlerProvider struct {
+	handler lambdaEventHandler
+}
+
+func (m mockHandlerProvider) getHandler(_ eventType) (lambdaEventHandler, error) {
+	return m.handler, nil
+}
+
 type mockPlogEventHandler struct {
 	handleCount int
 	event       eventType

receiver/awslambdareceiver/metadata.yaml

@@ -12,4 +12,4 @@ tests:
   # TODO Fix this
   skip_lifecycle: true
   config:
-    s3_encoding: awslogs
+    encoding_extension: awslogs