-
Notifications
You must be signed in to change notification settings - Fork 142
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Data Streams Monitoring support in Kinesis (#6428)
## Summary of changes * Added Data Streams Monitoring checkpoints to existing autoinstrumentation of Kinesis PutRecord[s][Async] * Added autoinstrumentation to Kinesis GetRecords[Async] ## Reason for change Data Streams Monitoring was not supported for Kinesis, this adds that functionality. ## Implementation details The existing producer instrumentation only works when StreamName is specified, not when StreamARN is specified. These changes leave that behavior unchanged at present. The new consumer instrumentation is able to infer a StreamName from the [mandatory] StreamARN so that the producer and consumer can use the same naming system. The reverse would not be possible, and changing producer instrumentation to use ARNs instead would be a potentially-breaking change to existing users of the functionality so is outside the scope of this feature addition.
- Loading branch information
Showing
23 changed files
with
1,608 additions
and
919 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
82 changes: 82 additions & 0 deletions
82
...c/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/GetRecordsAsyncIntegration.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| // <copyright file="GetRecordsAsyncIntegration.cs" company="Datadog"> | ||
| // Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License. | ||
| // This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc. | ||
| // </copyright> | ||
|
|
||
| #nullable enable | ||
|
|
||
| using System; | ||
| using System.ComponentModel; | ||
| using System.Threading; | ||
| using Datadog.Trace.ClrProfiler.AutoInstrumentation.AWS.Shared; | ||
| using Datadog.Trace.ClrProfiler.CallTarget; | ||
| using Datadog.Trace.DataStreamsMonitoring; | ||
| using Datadog.Trace.DuckTyping; | ||
| using Datadog.Trace.Propagators; | ||
|
|
||
| namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.AWS.Kinesis | ||
| { | ||
| /// <summary> | ||
| /// AWSSDK.Kinesis GetRecordsAsync CallTarget instrumentation | ||
| /// </summary> | ||
| [InstrumentMethod( | ||
| AssemblyName = "AWSSDK.Kinesis", | ||
| TypeName = "Amazon.Kinesis.AmazonKinesisClient", | ||
| MethodName = "GetRecordsAsync", | ||
| ReturnTypeName = "System.Threading.Tasks.Task`1[Amazon.Kinesis.Model.GetRecordsResponse]", | ||
| ParameterTypeNames = new[] { "Amazon.Kinesis.Model.GetRecordsRequest", ClrNames.CancellationToken }, | ||
| MinimumVersion = "3.0.0", | ||
| MaximumVersion = "3.*.*", | ||
| IntegrationName = AwsKinesisCommon.IntegrationName)] | ||
| [Browsable(false)] | ||
| [EditorBrowsable(EditorBrowsableState.Never)] | ||
| public class GetRecordsAsyncIntegration | ||
| { | ||
| private const string Operation = "GetRecords"; | ||
|
|
||
| internal static CallTargetState OnMethodBegin<TTarget, TGetRecordsRequest>(TTarget instance, TGetRecordsRequest request, CancellationToken cancellationToken) | ||
| where TGetRecordsRequest : IGetRecordsRequest, IDuckType | ||
| { | ||
| if (request.Instance is null) | ||
| { | ||
| return CallTargetState.GetDefault(); | ||
| } | ||
|
|
||
| var scope = AwsKinesisCommon.CreateScope(Tracer.Instance, Operation, SpanKinds.Consumer, null, out var tags); | ||
|
|
||
| string? streamName = AwsKinesisCommon.StreamNameFromARN(request.StreamARN); | ||
| if (tags is not null && streamName is not null) | ||
| { | ||
| tags.StreamName = streamName; | ||
| } | ||
|
|
||
| return new CallTargetState(scope, streamName); | ||
| } | ||
|
|
||
| internal static TResponse OnAsyncMethodEnd<TTarget, TResponse>(TTarget instance, TResponse response, Exception? exception, in CallTargetState state) | ||
| where TResponse : IGetRecordsResponse, IDuckType | ||
| { | ||
| if (response.Instance != null && response.Records is { Count: > 0 } && state is { State: not null, Scope.Span: { } span }) | ||
| { | ||
| var dataStreamsManager = Tracer.Instance.TracerManager.DataStreamsManager; | ||
| if (dataStreamsManager is { IsEnabled: true }) | ||
| { | ||
| var edgeTags = new[] { "direction:in", $"topic:{(string)state.State}", "type:kinesis" }; | ||
| foreach (var o in response.Records) | ||
| { | ||
| var record = o.DuckCast<IRecord>(); | ||
| if (record == null) | ||
| { | ||
| continue; // should not happen | ||
| } | ||
|
|
||
| span.SetDataStreamsCheckpoint(dataStreamsManager, CheckpointKind.Consume, edgeTags, payloadSizeBytes: 0, timeInQueueMs: 0); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| state.Scope.DisposeWithException(exception); | ||
| return response; | ||
| } | ||
| } | ||
| } |
99 changes: 99 additions & 0 deletions
99
...er/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/GetRecordsIntegration.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| // <copyright file="GetRecordsIntegration.cs" company="Datadog"> | ||
| // Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License. | ||
| // This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc. | ||
| // </copyright> | ||
|
|
||
| #nullable enable | ||
|
|
||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.ComponentModel; | ||
| using Datadog.Trace.ClrProfiler.CallTarget; | ||
| using Datadog.Trace.DataStreamsMonitoring; | ||
| using Datadog.Trace.DuckTyping; | ||
| using Datadog.Trace.Propagators; | ||
|
|
||
| namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.AWS.Kinesis | ||
| { | ||
| /// <summary> | ||
| /// AWSSDK.Kinesis GetRecords CallTarget instrumentation | ||
| /// </summary> | ||
| [InstrumentMethod( | ||
| AssemblyName = "AWSSDK.Kinesis", | ||
| TypeName = "Amazon.Kinesis.AmazonKinesisClient", | ||
| MethodName = "GetRecords", | ||
| ReturnTypeName = "Amazon.Kinesis.Model.GetRecordsResponse", | ||
| ParameterTypeNames = new[] { "Amazon.Kinesis.Model.GetRecordsRequest" }, | ||
| MinimumVersion = "3.0.0", | ||
| MaximumVersion = "3.*.*", | ||
| IntegrationName = AwsKinesisCommon.IntegrationName)] | ||
| [Browsable(false)] | ||
| [EditorBrowsable(EditorBrowsableState.Never)] | ||
| public class GetRecordsIntegration | ||
| { | ||
| private const string Operation = "GetRecords"; | ||
|
|
||
| /// <summary> | ||
| /// OnMethodBegin callback | ||
| /// </summary> | ||
| /// <typeparam name="TTarget">Type of the target</typeparam> | ||
| /// <typeparam name="TGetRecordsRequest">Type of the request object</typeparam> | ||
| /// <param name="instance">Instance value, aka `this` of the instrumented method</param> | ||
| /// <param name="request">The request for the Kinesis operation</param> | ||
| /// <returns>CallTarget state value</returns> | ||
| internal static CallTargetState OnMethodBegin<TTarget, TGetRecordsRequest>(TTarget instance, TGetRecordsRequest request) | ||
| where TGetRecordsRequest : IGetRecordsRequest, IDuckType | ||
| { | ||
| if (request.Instance is null) | ||
| { | ||
| return CallTargetState.GetDefault(); | ||
| } | ||
|
|
||
| var scope = AwsKinesisCommon.CreateScope(Tracer.Instance, Operation, SpanKinds.Producer, null, out var tags); | ||
|
|
||
| string? streamName = AwsKinesisCommon.StreamNameFromARN(request.StreamARN); | ||
| if (tags is not null && streamName is not null) | ||
| { | ||
| tags.StreamName = streamName; | ||
| } | ||
|
|
||
| return new CallTargetState(scope); | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// OnMethodEnd callback | ||
| /// </summary> | ||
| /// <typeparam name="TTarget">Type of the target</typeparam> | ||
| /// <typeparam name="TResponse">Type of the return value</typeparam> | ||
| /// <param name="instance">Instance value, aka `this` of the instrumented method.</param> | ||
| /// <param name="response">Task of HttpResponse message instance</param> | ||
| /// <param name="exception">Exception instance in case the original code threw an exception.</param> | ||
| /// <param name="state">Calltarget state value</param> | ||
| /// <returns>A response value, in an async scenario will be T of Task of T</returns> | ||
| internal static CallTargetReturn<TResponse> OnMethodEnd<TTarget, TResponse>(TTarget instance, TResponse response, Exception? exception, in CallTargetState state) | ||
| where TResponse : IGetRecordsResponse, IDuckType | ||
| { | ||
| if (response.Instance != null && response.Records is { Count: > 0 } && state is { State: not null, Scope.Span: { } span }) | ||
| { | ||
| var dataStreamsManager = Tracer.Instance.TracerManager.DataStreamsManager; | ||
| if (dataStreamsManager is { IsEnabled: true }) | ||
| { | ||
| var edgeTags = new[] { "direction:in", $"topic:{(string)state.State}", "type:kinesis" }; | ||
| foreach (var o in response.Records) | ||
| { | ||
| var record = o.DuckCast<IRecord>(); | ||
| if (record == null) | ||
| { | ||
| continue; // should not happen | ||
| } | ||
|
|
||
| span.SetDataStreamsCheckpoint(dataStreamsManager, CheckpointKind.Consume, edgeTags, payloadSizeBytes: 0, timeInQueueMs: 0); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| state.Scope.DisposeWithException(exception); | ||
| return new CallTargetReturn<TResponse>(response); | ||
| } | ||
| } | ||
| } |
19 changes: 19 additions & 0 deletions
19
tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/IGetRecordsRequest.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| // <copyright file="IGetRecordsRequest.cs" company="Datadog"> | ||
| // Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License. | ||
| // This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc. | ||
| // </copyright> | ||
|
|
||
| #nullable enable | ||
|
|
||
| using System.Collections; | ||
|
|
||
| namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.AWS.Kinesis | ||
| { | ||
| /// <summary> | ||
| /// GetRecordsRequest interface for duck typing. | ||
| /// </summary> | ||
| internal interface IGetRecordsRequest | ||
| { | ||
| string StreamARN { get; } | ||
| } | ||
| } |
28 changes: 28 additions & 0 deletions
28
tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/IGetRecordsResponse.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| // <copyright file="IGetRecordsResponse.cs" company="Datadog"> | ||
| // Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License. | ||
| // This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc. | ||
| // </copyright> | ||
|
|
||
| #nullable enable | ||
|
|
||
| using System.Collections; | ||
| using System.Collections.Generic; | ||
| using System.IO; | ||
| using Datadog.Trace.ClrProfiler.AutoInstrumentation.AWS.Shared; | ||
| using Datadog.Trace.DuckTyping; | ||
|
|
||
| namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.AWS.Kinesis | ||
| { | ||
| /// <summary> | ||
| /// GetRecordsRequest interface for duck typing. | ||
| /// </summary> | ||
| internal interface IGetRecordsResponse : IDuckType | ||
| { | ||
| IList Records { get; } // <IRecord> | ||
| } | ||
|
|
||
| internal interface IRecord | ||
| { | ||
| MemoryStream? Data { get; } | ||
| } | ||
| } |
Oops, something went wrong.