Skip to content

Commit

Permalink
Merge branch 'main' into rarguelloF/AIDM-326/fix-slog
Browse files Browse the repository at this point in the history
  • Loading branch information
rarguelloF authored Sep 12, 2024
2 parents e2adad9 + e081e4a commit 3f4869f
Show file tree
Hide file tree
Showing 7 changed files with 306 additions and 128 deletions.
31 changes: 23 additions & 8 deletions contrib/aws/aws-sdk-go-v2/aws/aws.go
Original file line number Diff line number Diff line change
Expand Up @@ -229,21 +229,23 @@ func tableName(requestInput middleware.InitializeInput) string {
func streamName(requestInput middleware.InitializeInput) string {
switch params := requestInput.Parameters.(type) {
case *kinesis.PutRecordInput:
return *params.StreamName
return coalesceNameOrArnResource(params.StreamName, params.StreamARN)
case *kinesis.PutRecordsInput:
return *params.StreamName
return coalesceNameOrArnResource(params.StreamName, params.StreamARN)
case *kinesis.AddTagsToStreamInput:
return *params.StreamName
return coalesceNameOrArnResource(params.StreamName, params.StreamARN)
case *kinesis.RemoveTagsFromStreamInput:
return *params.StreamName
return coalesceNameOrArnResource(params.StreamName, params.StreamARN)
case *kinesis.CreateStreamInput:
return *params.StreamName
if params.StreamName != nil {
return *params.StreamName
}
case *kinesis.DeleteStreamInput:
return *params.StreamName
return coalesceNameOrArnResource(params.StreamName, params.StreamARN)
case *kinesis.DescribeStreamInput:
return *params.StreamName
return coalesceNameOrArnResource(params.StreamName, params.StreamARN)
case *kinesis.DescribeStreamSummaryInput:
return *params.StreamName
return coalesceNameOrArnResource(params.StreamName, params.StreamARN)
case *kinesis.GetRecordsInput:
if params.StreamARN != nil {
streamArnValue := *params.StreamARN
Expand Down Expand Up @@ -353,3 +355,16 @@ func serviceName(cfg *config, awsService string) string {
defaultName := fmt.Sprintf("aws.%s", awsService)
return namingschema.ServiceNameOverrideV0(defaultName, defaultName)
}

func coalesceNameOrArnResource(name *string, arnVal *string) string {
if name != nil {
return *name
}

if arnVal != nil {
parts := strings.Split(*arnVal, "/")
return parts[len(parts)-1]
}

return ""
}
62 changes: 62 additions & 0 deletions contrib/aws/aws-sdk-go-v2/aws/aws_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
"github.com/aws/aws-sdk-go-v2/service/sns"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
"github.com/aws/smithy-go/middleware"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
Expand Down Expand Up @@ -1077,3 +1078,64 @@ func TestWithErrorCheck(t *testing.T) {
})
}
}

func TestStreamName(t *testing.T) {
dummyName := `my-stream`
dummyArn := `arn:aws:kinesis:us-east-1:111111111111:stream/` + dummyName

tests := []struct {
name string
input any
expected string
}{
{
name: "PutRecords with ARN",
input: &kinesis.PutRecordsInput{StreamARN: &dummyArn},
expected: dummyName,
},
{
name: "PutRecords with Name",
input: &kinesis.PutRecordsInput{StreamName: &dummyName},
expected: dummyName,
},
{
name: "PutRecords with both",
input: &kinesis.PutRecordsInput{StreamName: &dummyName, StreamARN: &dummyArn},
expected: dummyName,
},
{
name: "PutRecord with Name",
input: &kinesis.PutRecordInput{StreamName: &dummyName},
expected: dummyName,
},
{
name: "CreateStream",
input: &kinesis.CreateStreamInput{StreamName: &dummyName},
expected: dummyName,
},
{
name: "CreateStream with nothing",
input: &kinesis.CreateStreamInput{},
expected: "",
},
{
name: "GetRecords",
input: &kinesis.GetRecordsInput{StreamARN: &dummyArn},
expected: dummyName,
},
{
name: "GetRecords with nothing",
input: &kinesis.GetRecordsInput{},
expected: "",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
req := middleware.InitializeInput{
Parameters: tt.input,
}
val := streamName(req)
assert.Equal(t, tt.expected, val)
})
}
}
43 changes: 43 additions & 0 deletions contrib/cloud.google.com/go/pubsub.v1/internal/tracing/config.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
// Unless explicitly stated otherwise all files in this repository are licensed
// under the Apache License Version 2.0.
// This product includes software developed at Datadog (https://www.datadoghq.com/).
// Copyright 2024 Datadog, Inc.

package tracing

import (
"gopkg.in/DataDog/dd-trace-go.v1/internal/namingschema"
)

type config struct {
serviceName string
publishSpanName string
receiveSpanName string
measured bool
}

func defaultConfig() *config {
return &config{
serviceName: namingschema.ServiceNameOverrideV0("", ""),
publishSpanName: namingschema.OpName(namingschema.GCPPubSubOutbound),
receiveSpanName: namingschema.OpName(namingschema.GCPPubSubInbound),
measured: false,
}
}

// Option is used to customize spans started by WrapReceiveHandler or Publish.
type Option func(cfg *config)

// WithServiceName sets the service name tag for traces started by WrapReceiveHandler or Publish.
func WithServiceName(serviceName string) Option {
return func(cfg *config) {
cfg.serviceName = serviceName
}
}

// WithMeasured sets the measured tag for traces started by WrapReceiveHandler or Publish.
func WithMeasured() Option {
return func(cfg *config) {
cfg.measured = true
}
}
120 changes: 120 additions & 0 deletions contrib/cloud.google.com/go/pubsub.v1/internal/tracing/tracing.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
// Unless explicitly stated otherwise all files in this repository are licensed
// under the Apache License Version 2.0.
// This product includes software developed at Datadog (https://www.datadoghq.com/).
// Copyright 2024 Datadog, Inc.

package tracing

import (
"context"
"sync"
"time"

"gopkg.in/DataDog/dd-trace-go.v1/ddtrace"
"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/ext"
"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
"gopkg.in/DataDog/dd-trace-go.v1/internal/log"
"gopkg.in/DataDog/dd-trace-go.v1/internal/telemetry"
)

const componentName = "cloud.google.com/go/pubsub.v1"

func init() {
telemetry.LoadIntegration(componentName)
tracer.MarkIntegrationImported(componentName)
}

type Message struct {
ID string
Data []byte
OrderingKey string
Attributes map[string]string
DeliveryAttempt *int
PublishTime time.Time
}

type Topic interface {
String() string
}

type Subscription interface {
String() string
}

func TracePublish(ctx context.Context, topic Topic, msg *Message, opts ...Option) (context.Context, func(serverID string, err error)) {
cfg := defaultConfig()
for _, opt := range opts {
opt(cfg)
}
spanOpts := []ddtrace.StartSpanOption{
tracer.ResourceName(topic.String()),
tracer.SpanType(ext.SpanTypeMessageProducer),
tracer.Tag("message_size", len(msg.Data)),
tracer.Tag("ordering_key", msg.OrderingKey),
tracer.Tag(ext.Component, componentName),
tracer.Tag(ext.SpanKind, ext.SpanKindProducer),
tracer.Tag(ext.MessagingSystem, ext.MessagingSystemGCPPubsub),
}
if cfg.serviceName != "" {
spanOpts = append(spanOpts, tracer.ServiceName(cfg.serviceName))
}
if cfg.measured {
spanOpts = append(spanOpts, tracer.Measured())
}
span, ctx := tracer.StartSpanFromContext(
ctx,
cfg.publishSpanName,
spanOpts...,
)
if msg.Attributes == nil {
msg.Attributes = make(map[string]string)
}
if err := tracer.Inject(span.Context(), tracer.TextMapCarrier(msg.Attributes)); err != nil {
log.Debug("contrib/cloud.google.com/go/pubsub.v1/trace: failed injecting tracing attributes: %v", err)
}
span.SetTag("num_attributes", len(msg.Attributes))

var once sync.Once
closeSpan := func(serverID string, err error) {
once.Do(func() {
span.SetTag("server_id", serverID)
span.Finish(tracer.WithError(err))
})
}
return ctx, closeSpan
}

func TraceReceiveFunc(s Subscription, opts ...Option) func(ctx context.Context, msg *Message) (context.Context, func()) {
cfg := defaultConfig()
for _, opt := range opts {
opt(cfg)
}
log.Debug("contrib/cloud.google.com/go/pubsub.v1/trace: Wrapping Receive Handler: %#v", cfg)
return func(ctx context.Context, msg *Message) (context.Context, func()) {
parentSpanCtx, _ := tracer.Extract(tracer.TextMapCarrier(msg.Attributes))
opts := []ddtrace.StartSpanOption{
tracer.ResourceName(s.String()),
tracer.SpanType(ext.SpanTypeMessageConsumer),
tracer.Tag("message_size", len(msg.Data)),
tracer.Tag("num_attributes", len(msg.Attributes)),
tracer.Tag("ordering_key", msg.OrderingKey),
tracer.Tag("message_id", msg.ID),
tracer.Tag("publish_time", msg.PublishTime.String()),
tracer.Tag(ext.Component, componentName),
tracer.Tag(ext.SpanKind, ext.SpanKindConsumer),
tracer.Tag(ext.MessagingSystem, ext.MessagingSystemGCPPubsub),
tracer.ChildOf(parentSpanCtx),
}
if cfg.serviceName != "" {
opts = append(opts, tracer.ServiceName(cfg.serviceName))
}
if cfg.measured {
opts = append(opts, tracer.Measured())
}
span, ctx := tracer.StartSpanFromContext(ctx, cfg.receiveSpanName, opts...)
if msg.DeliveryAttempt != nil {
span.SetTag("delivery_attempt", *msg.DeliveryAttempt)
}
return ctx, func() { span.Finish() }
}
}
36 changes: 6 additions & 30 deletions contrib/cloud.google.com/go/pubsub.v1/option.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,41 +6,17 @@
package pubsub

import (
"gopkg.in/DataDog/dd-trace-go.v1/internal/namingschema"
"gopkg.in/DataDog/dd-trace-go.v1/contrib/cloud.google.com/go/pubsub.v1/internal/tracing"
)

type config struct {
serviceName string
publishSpanName string
receiveSpanName string
measured bool
}
// Option is used to customize spans started by WrapReceiveHandler or Publish.
type Option = tracing.Option

func defaultConfig() *config {
return &config{
serviceName: namingschema.ServiceNameOverrideV0("", ""),
publishSpanName: namingschema.OpName(namingschema.GCPPubSubOutbound),
receiveSpanName: namingschema.OpName(namingschema.GCPPubSubInbound),
measured: false,
}
}

// A Option is used to customize spans started by WrapReceiveHandler or Publish.
type Option func(cfg *config)

// A ReceiveOption has been deprecated in favor of Option.
// Deprecated: ReceiveOption has been deprecated in favor of Option.
type ReceiveOption = Option

// WithServiceName sets the service name tag for traces started by WrapReceiveHandler or Publish.
func WithServiceName(serviceName string) Option {
return func(cfg *config) {
cfg.serviceName = serviceName
}
}
var WithServiceName = tracing.WithServiceName

// WithMeasured sets the measured tag for traces started by WrapReceiveHandler or Publish.
func WithMeasured() Option {
return func(cfg *config) {
cfg.measured = true
}
}
var WithMeasured = tracing.WithMeasured
Loading

0 comments on commit 3f4869f

Please sign in to comment.