feat(tracing): export additional AWS fields (#318)

This commit injects attribute setters into our AWS instrumentation, such
that we can selectively export fields for a given input struct.

pkg/handler/forwarder/tracing/setter.go

@@ -0,0 +1,36 @@
+package tracing
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/smithy-go/middleware"
+	"go.opentelemetry.io/contrib/instrumentation/github.com/aws/aws-sdk-go-v2/otelaws"
+	"go.opentelemetry.io/otel/attribute"
+)
+
+var AttributeSetters = []otelaws.AttributeSetter{AttributeSetter}
+
+func AttributeSetter(_ context.Context, in middleware.InitializeInput) (attrs []attribute.KeyValue) {
+	// see https://opentelemetry.io/docs/specs/semconv/object-stores/s3/
+	switch v := in.Parameters.(type) {
+	case *s3.GetObjectInput:
+		attrs = append(attrs,
+			attribute.String("aws.s3.bucket", aws.ToString(v.Bucket)),
+			attribute.String("aws.s3.key", aws.ToString(v.Key)),
+		)
+	case *s3.CopyObjectInput:
+		attrs = append(attrs,
+			attribute.String("aws.s3.bucket", aws.ToString(v.Bucket)),
+			attribute.String("aws.s3.key", aws.ToString(v.Key)),
+			attribute.String("aws.s3.copy_source", aws.ToString(v.CopySource)),
+		)
+	case *s3.PutObjectInput:
+		attrs = append(attrs,
+			attribute.String("aws.s3.bucket", aws.ToString(v.Bucket)),
+			attribute.String("aws.s3.key", aws.ToString(v.Key)),
+		)
+	}
+	return
+}

pkg/handler/subscriber/tracing/setter.go

@@ -0,0 +1,43 @@
+package tracing
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
+	"github.com/aws/smithy-go/middleware"
+	"go.opentelemetry.io/contrib/instrumentation/github.com/aws/aws-sdk-go-v2/otelaws"
+	"go.opentelemetry.io/otel/attribute"
+)
+
+var AttributeSetters = []otelaws.AttributeSetter{AttributeSetter}
+
+func AttributeSetter(_ context.Context, in middleware.InitializeInput) (attrs []attribute.KeyValue) {
+	switch v := in.Parameters.(type) {
+	case *cloudwatchlogs.DescribeLogGroupsInput:
+		if s := v.LogGroupNamePattern; s != nil {
+			attrs = append(attrs, attribute.String("log_group_name_pattern", aws.ToString(s)))
+		}
+		if s := v.LogGroupNamePrefix; s != nil {
+			attrs = append(attrs, attribute.String("log_group_name_prefix", aws.ToString(s)))
+		}
+	case *cloudwatchlogs.DescribeSubscriptionFiltersInput:
+		attrs = append(attrs, attribute.String("log_group_name", aws.ToString(v.LogGroupName)))
+		if s := v.FilterNamePrefix; s != nil {
+			attrs = append(attrs, attribute.String("filter_name_prefix", aws.ToString(s)))
+		}
+	case *cloudwatchlogs.PutSubscriptionFilterInput:
+		attrs = append(attrs,
+			attribute.String("destination_arn", aws.ToString(v.DestinationArn)),
+			attribute.String("log_group_name", aws.ToString(v.LogGroupName)),
+			attribute.String("role_arn", aws.ToString(v.RoleArn)),
+			attribute.String("filter_name", aws.ToString(v.FilterName)),
+		)
+	case *cloudwatchlogs.DeleteSubscriptionFilterInput:
+		attrs = append(attrs,
+			attribute.String("log_group_name", aws.ToString(v.LogGroupName)),
+			attribute.String("filter_name", aws.ToString(v.FilterName)),
+		)
+	}
+	return
+}

pkg/lambda/forwarder/lambda.go

@@ -15,6 +15,7 @@ import (
 	"github.com/observeinc/aws-sam-apps/pkg/handler/forwarder"
 	"github.com/observeinc/aws-sam-apps/pkg/handler/forwarder/override"
 	"github.com/observeinc/aws-sam-apps/pkg/handler/forwarder/s3http"
+	forwardertracing "github.com/observeinc/aws-sam-apps/pkg/handler/forwarder/tracing"
 	"github.com/observeinc/aws-sam-apps/pkg/logging"
 	"github.com/observeinc/aws-sam-apps/pkg/tracing"
 	"github.com/observeinc/aws-sam-apps/pkg/version"
@@ -95,7 +96,11 @@ func New(ctx context.Context, cfg *Config) (*Lambda, error) {
 		span.End()
 	}()
 
-	awsCfg, err := tracing.AWSLoadDefaultConfig(ctx, tracerProvider)
+	awsCfg, err := tracing.AWSLoadDefaultConfig(ctx, &tracing.AWSConfig{
+		Logger:           logger,
+		TracerProvider:   tracerProvider,
+		AttributeSetters: forwardertracing.AttributeSetters,
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to load AWS configuration: %w", err)
 	}

pkg/lambda/subscriber/lambda.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/observeinc/aws-sam-apps/pkg/handler"
 	"github.com/observeinc/aws-sam-apps/pkg/handler/subscriber"
+	subscribertracing "github.com/observeinc/aws-sam-apps/pkg/handler/subscriber/tracing"
 	"github.com/observeinc/aws-sam-apps/pkg/logging"
 	"github.com/observeinc/aws-sam-apps/pkg/tracing"
 	"github.com/observeinc/aws-sam-apps/pkg/version"
@@ -85,7 +86,11 @@ func New(ctx context.Context, cfg *Config) (*Lambda, error) {
 		span.End()
 	}()
 
-	awsCfg, err := tracing.AWSLoadDefaultConfig(ctx, tracerProvider)
+	awsCfg, err := tracing.AWSLoadDefaultConfig(ctx, &tracing.AWSConfig{
+		Logger:           logger,
+		TracerProvider:   tracerProvider,
+		AttributeSetters: subscribertracing.AttributeSetters,
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to load AWS configuration: %w", err)
 	}

pkg/tracing/aws.go

@@ -6,26 +6,53 @@ import (
 	"os"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/aws/middleware"
+	v2middleware "github.com/aws/aws-sdk-go-v2/aws/middleware"
 	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/smithy-go/logging"
+	"github.com/go-logr/logr"
 	"go.opentelemetry.io/contrib/instrumentation/github.com/aws/aws-sdk-go-v2/otelaws"
 	"go.opentelemetry.io/otel/trace"
 
 	"github.com/observeinc/aws-sam-apps/pkg/version"
 )
 
-func AWSLoadDefaultConfig(ctx context.Context, tracerProvider trace.TracerProvider) (aws.Config, error) {
-	awsCfg, err := config.LoadDefaultConfig(ctx)
+type AWSConfig struct {
+	Logger           logr.Logger
+	TracerProvider   trace.TracerProvider
+	AttributeSetters []otelaws.AttributeSetter
+}
+
+func AWSLoadDefaultConfig(ctx context.Context, cfg *AWSConfig) (aws.Config, error) {
+	awsCfg, err := config.LoadDefaultConfig(ctx,
+		config.WithLogger(logging.LoggerFunc(func(classification logging.Classification, format string, v ...interface{}) {
+			switch classification {
+			case logging.Debug:
+				cfg.Logger.V(4).Info(fmt.Sprintf(format, v...))
+			case logging.Warn:
+				cfg.Logger.Info(fmt.Sprintf(format, v...))
+			}
+		})),
+	)
 	if err != nil {
 		return awsCfg, fmt.Errorf("failed to load AWS configuration: %w", err)
 	}
 
+	switch {
+	case cfg.Logger.V(8).Enabled():
+		awsCfg.ClientLogMode |= aws.LogRequestWithBody | aws.LogResponseWithBody
+	case cfg.Logger.V(6).Enabled():
+		awsCfg.ClientLogMode |= aws.LogRequest | aws.LogResponse
+	}
+
 	if serviceName := os.Getenv("OTEL_SERVICE_NAME"); serviceName != "" {
 		awsCfg.APIOptions = append(awsCfg.APIOptions,
-			middleware.AddUserAgentKeyValue(serviceName, version.Version),
+			v2middleware.AddUserAgentKeyValue(serviceName, version.Version),
 		)
 	}
 
-	otelaws.AppendMiddlewares(&awsCfg.APIOptions, otelaws.WithTracerProvider(tracerProvider))
+	otelaws.AppendMiddlewares(&awsCfg.APIOptions,
+		otelaws.WithTracerProvider(cfg.TracerProvider),
+		otelaws.WithAttributeSetter(append(cfg.AttributeSetters, otelaws.DefaultAttributeSetter)...),
+	)
 	return awsCfg, nil
 }