-
Notifications
You must be signed in to change notification settings - Fork 2.2k
/
segment.go
525 lines (463 loc) · 16.2 KB
/
segment.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0
package translator // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awsxrayexporter/internal/translator"
import (
"crypto/rand"
"encoding/binary"
"encoding/hex"
"encoding/json"
"fmt"
"net/url"
"regexp"
"strings"
"time"
awsP "github.com/aws/aws-sdk-go/aws"
"go.opentelemetry.io/collector/pdata/pcommon"
"go.opentelemetry.io/collector/pdata/ptrace"
conventions "go.opentelemetry.io/collector/semconv/v1.8.0"
awsxray "github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/xray"
"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/traceutil"
)
// AWS X-Ray acceptable values for origin field.
const (
OriginEC2 = "AWS::EC2::Instance"
OriginECS = "AWS::ECS::Container"
OriginECSEC2 = "AWS::ECS::EC2"
OriginECSFargate = "AWS::ECS::Fargate"
OriginEB = "AWS::ElasticBeanstalk::Environment"
OriginEKS = "AWS::EKS::Container"
OriginAppRunner = "AWS::AppRunner::Service"
)
// x-ray only span attributes - https://github.com/open-telemetry/opentelemetry-java-contrib/pull/802
const (
awsLocalService = "aws.local.service"
awsRemoteService = "aws.remote.service"
)
var (
// reInvalidSpanCharacters defines the invalid letters in a span name as per
// https://docs.aws.amazon.com/xray/latest/devguide/xray-api-segmentdocuments.html
reInvalidSpanCharacters = regexp.MustCompile(`[^ 0-9\p{L}N_.:/%&#=+,\-@]`)
)
const (
// defaultMetadataNamespace is used for non-namespaced non-indexed attributes.
defaultMetadataNamespace = "default"
// defaultSpanName will be used if there are no valid xray characters in the span name
defaultSegmentName = "span"
// maxSegmentNameLength the maximum length of a Segment name
maxSegmentNameLength = 200
// rpc.system value for AWS service remotes
awsAPIRPCSystem = "aws-api"
)
const (
traceIDLength = 35 // fixed length of aws trace id
identifierOffset = 11 // offset of identifier within traceID
)
var (
writers = newWriterPool(2048)
)
// MakeSegmentDocumentString converts an OpenTelemetry Span to an X-Ray Segment and then serialzies to JSON
func MakeSegmentDocumentString(span ptrace.Span, resource pcommon.Resource, indexedAttrs []string, indexAllAttrs bool, logGroupNames []string, skipTimestampValidation bool) (string, error) {
segment, err := MakeSegment(span, resource, indexedAttrs, indexAllAttrs, logGroupNames, skipTimestampValidation)
if err != nil {
return "", err
}
w := writers.borrow()
if err := w.Encode(*segment); err != nil {
return "", err
}
jsonStr := w.String()
writers.release(w)
return jsonStr, nil
}
func isAwsSdkSpan(span ptrace.Span) bool {
attributes := span.Attributes()
if rpcSystem, ok := attributes.Get(conventions.AttributeRPCSystem); ok {
return rpcSystem.Str() == awsAPIRPCSystem
}
return false
}
// MakeSegment converts an OpenTelemetry Span to an X-Ray Segment
func MakeSegment(span ptrace.Span, resource pcommon.Resource, indexedAttrs []string, indexAllAttrs bool, logGroupNames []string, skipTimestampValidation bool) (*awsxray.Segment, error) {
var segmentType string
storeResource := true
if span.Kind() != ptrace.SpanKindServer &&
!span.ParentSpanID().IsEmpty() {
segmentType = "subsegment"
// We only store the resource information for segments, the local root.
storeResource = false
}
// convert trace id
traceID, err := convertToAmazonTraceID(span.TraceID(), skipTimestampValidation)
if err != nil {
return nil, err
}
attributes := span.Attributes()
var (
startTime = timestampToFloatSeconds(span.StartTimestamp())
endTime = timestampToFloatSeconds(span.EndTimestamp())
httpfiltered, http = makeHTTP(span)
isError, isFault, isThrottle, causefiltered, cause = makeCause(span, httpfiltered, resource)
origin = determineAwsOrigin(resource)
awsfiltered, aws = makeAws(causefiltered, resource, logGroupNames)
service = makeService(resource)
sqlfiltered, sql = makeSQL(span, awsfiltered)
additionalAttrs = addSpecialAttributes(sqlfiltered, indexedAttrs, attributes)
user, annotations, metadata = makeXRayAttributes(additionalAttrs, resource, storeResource, indexedAttrs, indexAllAttrs)
spanLinks, makeSpanLinkErr = makeSpanLinks(span.Links(), skipTimestampValidation)
name string
namespace string
)
if makeSpanLinkErr != nil {
return nil, makeSpanLinkErr
}
// X-Ray segment names are service names, unlike span names which are methods. Try to find a service name.
// support x-ray specific service name attributes as segment name if it exists
if span.Kind() == ptrace.SpanKindServer || span.Kind() == ptrace.SpanKindConsumer {
if localServiceName, ok := attributes.Get(awsLocalService); ok {
name = localServiceName.Str()
}
}
if span.Kind() == ptrace.SpanKindClient || span.Kind() == ptrace.SpanKindProducer {
if remoteServiceName, ok := attributes.Get(awsRemoteService); ok {
name = remoteServiceName.Str()
// only strip the prefix for AWS spans
if isAwsSdkSpan(span) && strings.HasPrefix(name, "AWS.SDK.") {
name = strings.TrimPrefix(name, "AWS.SDK.")
}
}
}
// peer.service should always be prioritized for segment names when it set by users and
// the new x-ray specific service name attributes are not found
if name == "" {
if peerService, ok := attributes.Get(conventions.AttributePeerService); ok {
name = peerService.Str()
}
}
if namespace == "" {
if isAwsSdkSpan(span) {
namespace = conventions.AttributeCloudProviderAWS
}
}
if name == "" {
if awsService, ok := attributes.Get(awsxray.AWSServiceAttribute); ok {
// Generally spans are named something like "Method" or "Service.Method" but for AWS spans, X-Ray expects spans
// to be named "Service"
name = awsService.Str()
if namespace == "" {
namespace = conventions.AttributeCloudProviderAWS
}
}
}
if name == "" {
if dbInstance, ok := attributes.Get(conventions.AttributeDBName); ok {
// For database queries, the segment name convention is <db name>@<db host>
name = dbInstance.Str()
if dbURL, ok := attributes.Get(conventions.AttributeDBConnectionString); ok {
if parsed, _ := url.Parse(dbURL.Str()); parsed != nil {
if parsed.Hostname() != "" {
name += "@" + parsed.Hostname()
}
}
}
}
}
if name == "" && span.Kind() == ptrace.SpanKindServer {
// Only for a server span, we can use the resource.
if service, ok := resource.Attributes().Get(conventions.AttributeServiceName); ok {
name = service.Str()
}
}
if name == "" {
if rpcservice, ok := attributes.Get(conventions.AttributeRPCService); ok {
name = rpcservice.Str()
}
}
if name == "" {
if host, ok := attributes.Get(conventions.AttributeHTTPHost); ok {
name = host.Str()
}
}
if name == "" {
if peer, ok := attributes.Get(conventions.AttributeNetPeerName); ok {
name = peer.Str()
}
}
if name == "" {
name = fixSegmentName(span.Name())
}
if namespace == "" && span.Kind() == ptrace.SpanKindClient {
namespace = "remote"
}
return &awsxray.Segment{
ID: awsxray.String(traceutil.SpanIDToHexOrEmptyString(span.SpanID())),
TraceID: awsxray.String(traceID),
Name: awsxray.String(name),
StartTime: awsP.Float64(startTime),
EndTime: awsP.Float64(endTime),
ParentID: awsxray.String(traceutil.SpanIDToHexOrEmptyString(span.ParentSpanID())),
Fault: awsP.Bool(isFault),
Error: awsP.Bool(isError),
Throttle: awsP.Bool(isThrottle),
Cause: cause,
Origin: awsxray.String(origin),
Namespace: awsxray.String(namespace),
User: awsxray.String(user),
HTTP: http,
AWS: aws,
Service: service,
SQL: sql,
Annotations: annotations,
Metadata: metadata,
Type: awsxray.String(segmentType),
Links: spanLinks,
}, nil
}
// newSegmentID generates a new valid X-Ray SegmentID
func newSegmentID() pcommon.SpanID {
var r [8]byte
_, err := rand.Read(r[:])
if err != nil {
panic(err)
}
return pcommon.SpanID(r)
}
func determineAwsOrigin(resource pcommon.Resource) string {
if resource.Attributes().Len() == 0 {
return ""
}
if provider, ok := resource.Attributes().Get(conventions.AttributeCloudProvider); ok {
if provider.Str() != conventions.AttributeCloudProviderAWS {
return ""
}
}
if is, present := resource.Attributes().Get(conventions.AttributeCloudPlatform); present {
switch is.Str() {
case conventions.AttributeCloudPlatformAWSAppRunner:
return OriginAppRunner
case conventions.AttributeCloudPlatformAWSEKS:
return OriginEKS
case conventions.AttributeCloudPlatformAWSElasticBeanstalk:
return OriginEB
case conventions.AttributeCloudPlatformAWSECS:
lt, present := resource.Attributes().Get(conventions.AttributeAWSECSLaunchtype)
if !present {
return OriginECS
}
switch lt.Str() {
case conventions.AttributeAWSECSLaunchtypeEC2:
return OriginECSEC2
case conventions.AttributeAWSECSLaunchtypeFargate:
return OriginECSFargate
default:
return OriginECS
}
case conventions.AttributeCloudPlatformAWSEC2:
return OriginEC2
// If cloud_platform is defined with a non-AWS value, we should not assign it an AWS origin
default:
return ""
}
}
return ""
}
// convertToAmazonTraceID converts a trace ID to the Amazon format.
//
// A trace ID unique identifier that connects all segments and subsegments
// originating from a single client request.
// - A trace_id consists of three numbers separated by hyphens. For example,
// 1-58406520-a006649127e371903a2de979. This includes:
// - The version number, that is, 1.
// - The time of the original request, in Unix epoch time, in 8 hexadecimal digits.
// - For example, 10:00AM December 2nd, 2016 PST in epoch time is 1480615200 seconds,
// or 58406520 in hexadecimal.
// - A 96-bit identifier for the trace, globally unique, in 24 hexadecimal digits.
func convertToAmazonTraceID(traceID pcommon.TraceID, skipTimestampValidation bool) (string, error) {
const (
// maxAge of 28 days. AWS has a 30 day limit, let's be conservative rather than
// hit the limit
maxAge = 60 * 60 * 24 * 28
// maxSkew allows for 5m of clock skew
maxSkew = 60 * 5
)
var (
content = [traceIDLength]byte{}
epochNow = time.Now().Unix()
traceIDBytes = traceID
epoch = int64(binary.BigEndian.Uint32(traceIDBytes[0:4]))
b = [4]byte{}
)
// If feature gate is enabled, skip the timestamp validation logic
if !skipTimestampValidation {
// If AWS traceID originally came from AWS, no problem. However, if oc generated
// the traceID, then the epoch may be outside the accepted AWS range of within the
// past 30 days.
//
// In that case, we return invalid traceid error
if delta := epochNow - epoch; delta > maxAge || delta < -maxSkew {
return "", fmt.Errorf("invalid xray traceid: %s", traceID)
}
}
binary.BigEndian.PutUint32(b[0:4], uint32(epoch))
content[0] = '1'
content[1] = '-'
hex.Encode(content[2:10], b[0:4])
content[10] = '-'
hex.Encode(content[identifierOffset:], traceIDBytes[4:16]) // overwrite with identifier
return string(content[0:traceIDLength]), nil
}
func timestampToFloatSeconds(ts pcommon.Timestamp) float64 {
return float64(ts) / float64(time.Second)
}
func addSpecialAttributes(attributes map[string]pcommon.Value, indexedAttrs []string, unfilteredAttributes pcommon.Map) map[string]pcommon.Value {
for _, name := range indexedAttrs {
// Allow attributes that have been filtered out before if explicitly added to be annotated/indexed
_, isAnnotatable := attributes[name]
unfilteredAttribute, attributeExists := unfilteredAttributes.Get(name)
if !isAnnotatable && attributeExists {
attributes[name] = unfilteredAttribute
}
}
return attributes
}
func makeXRayAttributes(attributes map[string]pcommon.Value, resource pcommon.Resource, storeResource bool, indexedAttrs []string, indexAllAttrs bool) (
string, map[string]any, map[string]map[string]any) {
var (
annotations = map[string]any{}
metadata = map[string]map[string]any{}
user string
)
userid, ok := attributes[conventions.AttributeEnduserID]
if ok {
user = userid.Str()
delete(attributes, conventions.AttributeEnduserID)
}
if len(attributes) == 0 && (!storeResource || resource.Attributes().Len() == 0) {
return user, nil, nil
}
defaultMetadata := map[string]any{}
indexedKeys := map[string]bool{}
if !indexAllAttrs {
for _, name := range indexedAttrs {
indexedKeys[name] = true
}
}
annotationKeys, ok := attributes[awsxray.AWSXraySegmentAnnotationsAttribute]
if ok && annotationKeys.Type() == pcommon.ValueTypeSlice {
slice := annotationKeys.Slice()
for i := 0; i < slice.Len(); i++ {
value := slice.At(i)
if value.Type() != pcommon.ValueTypeStr {
continue
}
key := value.AsString()
indexedKeys[key] = true
}
delete(attributes, awsxray.AWSXraySegmentAnnotationsAttribute)
}
if storeResource {
resource.Attributes().Range(func(key string, value pcommon.Value) bool {
key = "otel.resource." + key
annoVal := annotationValue(value)
indexed := indexAllAttrs || indexedKeys[key]
if annoVal != nil && indexed {
key = fixAnnotationKey(key)
annotations[key] = annoVal
} else {
metaVal := value.AsRaw()
if metaVal != nil {
defaultMetadata[key] = metaVal
}
}
return true
})
}
if indexAllAttrs {
for key, value := range attributes {
key = fixAnnotationKey(key)
annoVal := annotationValue(value)
if annoVal != nil {
annotations[key] = annoVal
}
}
} else {
for key, value := range attributes {
switch {
case indexedKeys[key]:
key = fixAnnotationKey(key)
annoVal := annotationValue(value)
if annoVal != nil {
annotations[key] = annoVal
}
case strings.HasPrefix(key, awsxray.AWSXraySegmentMetadataAttributePrefix) && value.Type() == pcommon.ValueTypeStr:
namespace := strings.TrimPrefix(key, awsxray.AWSXraySegmentMetadataAttributePrefix)
var metaVal map[string]any
err := json.Unmarshal([]byte(value.Str()), &metaVal)
switch {
case err != nil:
// if unable to unmarshal, keep the original key/value
defaultMetadata[key] = value.Str()
case strings.EqualFold(namespace, defaultMetadataNamespace):
for k, v := range metaVal {
defaultMetadata[k] = v
}
default:
metadata[namespace] = metaVal
}
default:
metaVal := value.AsRaw()
if metaVal != nil {
defaultMetadata[key] = metaVal
}
}
}
}
if len(defaultMetadata) > 0 {
metadata[defaultMetadataNamespace] = defaultMetadata
}
return user, annotations, metadata
}
func annotationValue(value pcommon.Value) any {
switch value.Type() {
case pcommon.ValueTypeStr:
return value.Str()
case pcommon.ValueTypeInt:
return value.Int()
case pcommon.ValueTypeDouble:
return value.Double()
case pcommon.ValueTypeBool:
return value.Bool()
}
return nil
}
// fixSegmentName removes any invalid characters from the span name. AWS X-Ray defines
// the list of valid characters here:
// https://docs.aws.amazon.com/xray/latest/devguide/xray-api-segmentdocuments.html
func fixSegmentName(name string) string {
if reInvalidSpanCharacters.MatchString(name) {
// only allocate for ReplaceAllString if we need to
name = reInvalidSpanCharacters.ReplaceAllString(name, "")
}
if length := len(name); length > maxSegmentNameLength {
name = name[0:maxSegmentNameLength]
} else if length == 0 {
name = defaultSegmentName
}
return name
}
// fixAnnotationKey removes any invalid characters from the annotaiton key. AWS X-Ray defines
// the list of valid characters here:
// https://docs.aws.amazon.com/xray/latest/devguide/xray-api-segmentdocuments.html
func fixAnnotationKey(key string) string {
return strings.Map(func(r rune) rune {
switch {
case '0' <= r && r <= '9':
fallthrough
case 'A' <= r && r <= 'Z':
fallthrough
case 'a' <= r && r <= 'z':
return r
default:
return '_'
}
}, key)
}