otelaws: adding dynamodb attributes

CHANGELOG.md

@@ -9,6 +9,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ## [Unreleased]
 
 ### Added
+- Dynamodb spans will now have the appropriate attributes added for the operation being performed, this is detected automatically but it is also now possible to provide a custom function to set attributes using `WithAttributeSetter` 
 
 - Add `WithClientTrace` option to `otelhttp.Transport` (#875)
 

instrumentation/github.com/aws/aws-sdk-go-v2/otelaws/attributes.go

@@ -14,7 +14,14 @@
 
 package otelaws
 
-import "go.opentelemetry.io/otel/attribute"
+import (
+	"context"
+
+	v2Middleware "github.com/aws/aws-sdk-go-v2/aws/middleware"
+	"github.com/aws/smithy-go/middleware"
+
+	"go.opentelemetry.io/otel/attribute"
+)
 
 // AWS attributes.
 const (
@@ -24,6 +31,10 @@ const (
 	RequestIDKey attribute.Key = "aws.request_id"
 )
 
+var servicemap map[string]AttributeSetter = map[string]AttributeSetter{
+	"dynamodb": DynamodbAttributeSetter,
+}
+
 func OperationAttr(operation string) attribute.KeyValue {
 	return OperationKey.String(operation)
 }
@@ -39,3 +50,14 @@ func ServiceAttr(service string) attribute.KeyValue {
 func RequestIDAttr(requestID string) attribute.KeyValue {
 	return RequestIDKey.String(requestID)
 }
+
+func DefaultAttributeSetter(ctx context.Context, in middleware.InitializeInput) []attribute.KeyValue {
+
+	serviceID := v2Middleware.GetServiceID(ctx)
+
+	if fn, ok := servicemap[serviceID]; ok {
+		return fn(ctx, in)
+	}
+
+	return []attribute.KeyValue{}
+}

instrumentation/github.com/aws/aws-sdk-go-v2/otelaws/aws.go

@@ -23,6 +23,7 @@ import (
 	smithyhttp "github.com/aws/smithy-go/transport/http"
 
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
 	semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
 	"go.opentelemetry.io/otel/trace"
@@ -34,8 +35,11 @@ const (
 
 type spanTimestampKey struct{}
 
+type AttributeSetter func(context.Context, middleware.InitializeInput) []attribute.KeyValue
+
 type otelMiddlewares struct {
-	tracer trace.Tracer
+	tracer          trace.Tracer
+	attributesetter []AttributeSetter
 }
 
 func (m otelMiddlewares) initializeMiddlewareBefore(stack *middleware.Stack) error {
@@ -55,12 +59,19 @@ func (m otelMiddlewares) initializeMiddlewareAfter(stack *middleware.Stack) erro
 		out middleware.InitializeOutput, metadata middleware.Metadata, err error) {
 
 		serviceID := v2Middleware.GetServiceID(ctx)
+
+		attributes := []attribute.KeyValue{ServiceAttr(serviceID),
+			RegionAttr(v2Middleware.GetRegion(ctx)),
+			OperationAttr(v2Middleware.GetOperationName(ctx)),
+		}
+		for _, setter := range m.attributesetter {
+			attributes = append(attributes, setter(ctx, in)...)
+		}
+
 		ctx, span := m.tracer.Start(ctx, serviceID,
 			trace.WithTimestamp(ctx.Value(spanTimestampKey{}).(time.Time)),
 			trace.WithSpanKind(trace.SpanKindClient),
-			trace.WithAttributes(ServiceAttr(serviceID),
-				RegionAttr(v2Middleware.GetRegion(ctx)),
-				OperationAttr(v2Middleware.GetOperationName(ctx))),
+			trace.WithAttributes(attributes...),
 		)
 		defer span.End()
 
@@ -110,7 +121,13 @@ func AppendMiddlewares(apiOptions *[]func(*middleware.Stack) error, opts ...Opti
 		opt.apply(&cfg)
 	}
 
+	if cfg.AttributeSetter == nil {
+		cfg.AttributeSetter = []AttributeSetter{DefaultAttributeSetter}
+	}
+
 	m := otelMiddlewares{tracer: cfg.TracerProvider.Tracer(tracerName,
-		trace.WithInstrumentationVersion(SemVersion()))}
+		trace.WithInstrumentationVersion(SemVersion())),
+		attributesetter: cfg.AttributeSetter}
 	*apiOptions = append(*apiOptions, m.initializeMiddlewareBefore, m.initializeMiddlewareAfter, m.deserializeMiddleware)
+
 }

instrumentation/github.com/aws/aws-sdk-go-v2/otelaws/config.go

@@ -19,7 +19,8 @@ import (
 )
 
 type config struct {
-	TracerProvider trace.TracerProvider
+	TracerProvider  trace.TracerProvider
+	AttributeSetter []AttributeSetter
 }
 
 // Option applies an option value.
@@ -44,3 +45,14 @@ func WithTracerProvider(provider trace.TracerProvider) Option {
 		}
 	})
 }
+
+// WithAttributeSetter specifies an attribute setter function for setting service specific attributes.
+// If none is specified, the service will be determined by the DefaultAttributeSetter function and the corresponding attributes will be included.
+func WithAttributeSetter(attributesetter []AttributeSetter) Option {
+
+	return optionFunc(func(cfg *config) {
+		if attributesetter != nil {
+			cfg.AttributeSetter = attributesetter
+		}
+	})
+}

instrumentation/github.com/aws/aws-sdk-go-v2/otelaws/dynamodbattributes.go

@@ -0,0 +1,188 @@
+// 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.
+
+package otelaws
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	"github.com/aws/smithy-go/middleware"
+
+	"go.opentelemetry.io/otel/attribute"
+	semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
+)
+
+func DynamodbAttributeSetter(ctx context.Context, in middleware.InitializeInput) []attribute.KeyValue {
+
+	dynamodbAttributes := []attribute.KeyValue{
+		{
+			Key:   semconv.DBSystemKey,
+			Value: attribute.StringValue("dynamodb"),
+		},
+	}
+
+	switch v := in.Parameters.(type) {
+	case *dynamodb.GetItemInput:
+
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+		if v.ConsistentRead != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBConsistentReadKey.Bool(*v.ConsistentRead))
+		}
+
+		if v.ProjectionExpression != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBProjectionKey.String(*v.ProjectionExpression))
+		}
+
+	case *dynamodb.BatchGetItemInput:
+		var tableNames []string
+		for k, _ := range v.RequestItems {
+			tableNames = append(tableNames, k)
+		}
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.StringSlice(tableNames))
+
+	case *dynamodb.BatchWriteItemInput:
+		var tableNames []string
+		for k, _ := range v.RequestItems {
+			tableNames = append(tableNames, k)
+		}
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.StringSlice(tableNames))
+
+	case *dynamodb.CreateTableInput:
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+		if v.GlobalSecondaryIndexes != nil {
+			globalindexes, _ := json.Marshal(v.GlobalSecondaryIndexes)
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBGlobalSecondaryIndexesKey.String(string(globalindexes)))
+		}
+
+		if v.LocalSecondaryIndexes != nil {
+			localindexes, _ := json.Marshal(v.LocalSecondaryIndexes)
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBLocalSecondaryIndexesKey.String(string(localindexes)))
+		}
+
+		if v.ProvisionedThroughput != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBProvisionedReadCapacityKey.Int64(*v.ProvisionedThroughput.ReadCapacityUnits))
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBProvisionedWriteCapacityKey.Int64(*v.ProvisionedThroughput.WriteCapacityUnits))
+		}
+
+	case *dynamodb.DeleteItemInput:
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+	case *dynamodb.DeleteTableInput:
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+	case *dynamodb.DescribeTableInput:
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+	case *dynamodb.ListTablesInput:
+
+		if v.ExclusiveStartTableName != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBExclusiveStartTableKey.String(*v.ExclusiveStartTableName))
+		}
+
+		if v.Limit != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBLimitKey.Int(int(*v.Limit)))
+		}
+
+	case *dynamodb.PutItemInput:
+
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+	case *dynamodb.QueryInput:
+
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+		if v.ConsistentRead != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBConsistentReadKey.Bool(*v.ConsistentRead))
+		}
+
+		if v.IndexName != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBIndexNameKey.String(*v.IndexName))
+		}
+
+		if v.Limit != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBLimitKey.Int(int(*v.Limit)))
+		}
+
+		if v.ScanIndexForward != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBScanForwardKey.Bool(*v.ScanIndexForward))
+		}
+
+		if v.ProjectionExpression != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBProjectionKey.String(*v.ProjectionExpression))
+		}
+
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBSelectKey.String(string(v.Select)))
+
+	case *dynamodb.ScanInput:
+
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+		if v.ConsistentRead != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBConsistentReadKey.Bool(*v.ConsistentRead))
+		}
+
+		if v.IndexName != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBIndexNameKey.String(*v.IndexName))
+		}
+
+		if v.Limit != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBLimitKey.Int(int(*v.Limit)))
+		}
+
+		if v.ProjectionExpression != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBProjectionKey.String(*v.ProjectionExpression))
+		}
+
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBSelectKey.String(string(v.Select)))
+
+		if v.Segment != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBSegmentKey.Int(int(*v.Segment)))
+		}
+
+		if v.TotalSegments != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTotalSegmentsKey.Int(int(*v.TotalSegments)))
+		}
+
+	case *dynamodb.UpdateItemInput:
+
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+	case *dynamodb.UpdateTableInput:
+
+		dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBTableNamesKey.String(*v.TableName))
+
+		if v.AttributeDefinitions != nil {
+			attributedefinitions, _ := json.Marshal(v.AttributeDefinitions)
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBAttributeDefinitionsKey.String(string(attributedefinitions)))
+		}
+
+		if v.GlobalSecondaryIndexUpdates != nil {
+			globalsecondaryindexupdates, _ := json.Marshal(v.GlobalSecondaryIndexUpdates)
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBGlobalSecondaryIndexUpdatesKey.String(string(globalsecondaryindexupdates)))
+		}
+
+		if v.ProvisionedThroughput != nil {
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBProvisionedReadCapacityKey.Int64(*v.ProvisionedThroughput.ReadCapacityUnits))
+			dynamodbAttributes = append(dynamodbAttributes, semconv.AWSDynamoDBProvisionedWriteCapacityKey.Int64(*v.ProvisionedThroughput.WriteCapacityUnits))
+		}
+
+	}
+
+	return dynamodbAttributes
+
+}