feat: auto-inject AWS account ID into telemetry from Lambda extension registration

Automatically extract the AWS account ID from the Lambda Extensions API
registration response and inject it as the 'cloud.account.id' attribute
into all telemetry (traces, logs, metrics) via a confmap converter.

Changes:
- Add AccountId field to RegisterResponse struct
- Request Lambda-Extension-Accept-Feature header with 'accountId' value
- Create accountidprocessor converter to inject cloud.account.id attribute
- Update lambdacomponents.Components to accept accountID and return converters
- Update collector.NewCollector to accept and register custom converters
- Update manager to pass account ID through the initialization flow

Includes:
- Comprehensive tests for JSON unmarshaling with leading zero preservation
- Tests for converter behavior across different pipeline configurations
- Tests for account ID handling in the extension API client

Benefits:
- Account ID automatically available in all telemetry without configuration
- No environment variables needed, uses AWS Lambda API response
- Follows OpenTelemetry Collector converter pattern
- Static injection at startup, no runtime overhead

Author: RaphaelManke

collector/internal/collector/collector.go

@@ -69,17 +69,22 @@ func getConfig(logger *zap.Logger) string {
 	return defaultVal
 }
 
-func NewCollector(logger *zap.Logger, factories otelcol.Factories, version string) *Collector {
+func NewCollector(logger *zap.Logger, factories otelcol.Factories, version string, customConverters []confmap.ConverterFactory) *Collector {
 	l := logger.Named("NewCollector")
+
+	// Combine built-in converters with custom converters
+	converters := []confmap.ConverterFactory{
+		confmap.NewConverterFactory(func(set confmap.ConverterSettings) confmap.Converter {
+			return disablequeuedretryconverter.New()
+		}),
+	}
+	converters = append(converters, customConverters...)
+
 	cfgSet := otelcol.ConfigProviderSettings{
 		ResolverSettings: confmap.ResolverSettings{
 			URIs:              []string{getConfig(l)},
 			ProviderFactories: []confmap.ProviderFactory{fileprovider.NewFactory(), envprovider.NewFactory(), yamlprovider.NewFactory(), httpsprovider.NewFactory(), httpprovider.NewFactory(), s3provider.NewFactory(), secretsmanagerprovider.NewFactory()},
-			ConverterFactories: []confmap.ConverterFactory{
-				confmap.NewConverterFactory(func(set confmap.ConverterSettings) confmap.Converter {
-					return disablequeuedretryconverter.New()
-				}),
-			},
+			ConverterFactories: converters,
 		},
 	}
 

collector/internal/confmap/converter/accountidprocessor/converter.go

@@ -0,0 +1,103 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// The accountidprocessor implements the Converter for mutating Collector
+// configurations to automatically inject the cloud.account.id attribute
+// via a resource processor into all pipelines.
+package accountidprocessor
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/collector/confmap"
+)
+
+const (
+	serviceKey       = "service"
+	pipelinesKey     = "pipelines"
+	processorsKey    = "processors"
+	resourceProc     = "resource/aws-account-id"
+	accountIDAttrKey = "cloud.account.id"
+)
+
+type converter struct {
+	accountID string
+}
+
+// New returns a confmap.Converter that injects cloud.account.id into all pipelines
+func New(accountID string) confmap.Converter {
+	return &converter{accountID: accountID}
+}
+
+func (c converter) Convert(_ context.Context, conf *confmap.Conf) error {
+	if c.accountID == "" {
+		return nil // Skip if no account ID
+	}
+
+	// Navigate to service.pipelines
+	serviceVal := conf.Get(serviceKey)
+	service, ok := serviceVal.(map[string]interface{})
+	if !ok {
+		return nil
+	}
+
+	pipelinesVal, ok := service[pipelinesKey]
+	if !ok {
+		return nil
+	}
+
+	pipelines, ok := pipelinesVal.(map[string]interface{})
+	if !ok {
+		return nil
+	}
+
+	updates := make(map[string]interface{})
+
+	// For each pipeline, add resource processor to beginning
+	for telemetryType, pipelineVal := range pipelines {
+		pipeline, ok := pipelineVal.(map[string]interface{})
+		if !ok {
+			continue
+		}
+
+		processorsVal, _ := pipeline[processorsKey]
+		processors, ok := processorsVal.([]interface{})
+		if !ok {
+			processors = []interface{}{}
+		}
+
+		// Prepend resource/aws-account-id processor
+		processors = append([]interface{}{resourceProc}, processors...)
+		updates[fmt.Sprintf("%s::%s::%s::%s", serviceKey, pipelinesKey, telemetryType, processorsKey)] = processors
+	}
+
+	// Configure the resource processor with cloud.account.id attribute
+	updates[fmt.Sprintf("processors::%s::attributes", resourceProc)] = []map[string]interface{}{
+		{
+			"key":    accountIDAttrKey,
+			"value":  c.accountID,
+			"action": "insert",
+		},
+	}
+
+	// Apply all updates
+	if len(updates) > 0 {
+		if err := conf.Merge(confmap.NewFromStringMap(updates)); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}