#1034: AWS SQS and SNS support

src/OpenTelemetry.Contrib.Instrumentation.AWS/Implementation/AWSMessagingUtils.cs

@@ -0,0 +1,66 @@
+// <copyright file="AWSMessagingUtils.cs" company="OpenTelemetry Authors">
+// 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.
+// </copyright>
+
+using System.Collections.Generic;
+using OpenTelemetry.Context.Propagation;
+
+namespace OpenTelemetry.Contrib.Instrumentation.AWS.Implementation;
+
+internal static class AWSMessagingUtils
+{
+    // SQS/SNS message attributes collection size limit according to
+    // https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-message-metadata.html and
+    // https://docs.aws.amazon.com/sns/latest/dg/sns-message-attributes.html
+    private const int MaxMessageAttributes = 10;
+
+    internal static void Inject(IRequestContextAdapter requestAdapter, PropagationContext propagationContext)
+    {
+        if (!requestAdapter.HasMessageBody ||
+            !requestAdapter.HasOriginalRequest)
+        {
+            return;
+        }
+
+        var carrier = new Dictionary<string, string>();
+        Propagators.DefaultTextMapPropagator.Inject(propagationContext, carrier, Setter);
+
+        int attributesCount = requestAdapter.AttributesCount;
+        if (carrier.Count + attributesCount > MaxMessageAttributes)
+        {
+            // TODO: Add logging (event source).
+            return;
+        }
+
+        int nextAttributeIndex = attributesCount + 1;
+        foreach (var param in carrier)
+        {
+            if (requestAdapter.ContainsAttribute(param.Key))
+            {
+                continue;
+            }
+
+            requestAdapter.AddAttribute(param.Key, param.Value, nextAttributeIndex);
+            nextAttributeIndex++;
+
+            // Add trace data to message attributes dictionary of the original request.
+            // This dictionary must be in sync with parameters collection to pass through the MD5 hash matching check.
+            requestAdapter.AddAttributeToOriginalRequest(param.Key, param.Value);
+        }
+    }
+
+    private static void Setter(IDictionary<string, string> carrier, string name, string value) =>
+        carrier[name] = value;
+}

src/OpenTelemetry.Contrib.Instrumentation.AWS/Implementation/AWSServiceHelper.cs

@@ -14,7 +14,6 @@
 // limitations under the License.
 // </copyright>
 
-using System;
 using System.Collections.Generic;
 using Amazon.Runtime;
 
@@ -24,8 +23,8 @@ internal class AWSServiceHelper
 {
     internal static IReadOnlyDictionary<string, string> ServiceParameterMap = new Dictionary<string, string>()
     {
-        { DynamoDbService, "TableName" },
-        { SQSService, "QueueUrl" },
+        { AWSServiceType.DynamoDbService, "TableName" },
+        { AWSServiceType.SQSService, "QueueUrl" },
     };
 
     internal static IReadOnlyDictionary<string, string> ParameterAttributeMap = new Dictionary<string, string>()
@@ -34,9 +33,6 @@ internal class AWSServiceHelper
         { "QueueUrl", AWSSemanticConventions.AttributeAWSSQSQueueUrl },
     };
 
-    private const string DynamoDbService = "DynamoDBv2";
-    private const string SQSService = "SQS";
-
     internal static string GetAWSServiceName(IRequestContext requestContext)
         => Utils.RemoveAmazonPrefixFromServiceName(requestContext.Request.ServiceName);
 
@@ -47,7 +43,4 @@ internal static string GetAWSOperationName(IRequestContext requestContext)
         var operationName = Utils.RemoveSuffix(completeRequestName, suffix);
         return operationName;
     }
-
-    internal static bool IsDynamoDbService(string service)
-        => DynamoDbService.Equals(service, StringComparison.OrdinalIgnoreCase);
 }

src/OpenTelemetry.Contrib.Instrumentation.AWS/Implementation/AWSServiceType.cs

@@ -0,0 +1,35 @@
+// <copyright file="AWSServiceType.cs" company="OpenTelemetry Authors">
+// 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.
+// </copyright>
+
+using System;
+
+namespace OpenTelemetry.Contrib.Instrumentation.AWS.Implementation;
+
+internal class AWSServiceType
+{
+    internal const string DynamoDbService = "DynamoDBv2";
+    internal const string SQSService = "SQS";
+    internal const string SNSService = "SimpleNotificationService"; // SNS
+
+    internal static bool IsDynamoDbService(string service)
+        => DynamoDbService.Equals(service, StringComparison.OrdinalIgnoreCase);
+
+    internal static bool IsSqsService(string service)
+        => SQSService.Equals(service, StringComparison.OrdinalIgnoreCase);
+
+    internal static bool IsSnsService(string service)
+        => SNSService.Equals(service, StringComparison.OrdinalIgnoreCase);
+}

src/OpenTelemetry.Contrib.Instrumentation.AWS/Implementation/AWSTracingPipelineHandler.cs

@@ -196,10 +196,18 @@ private void AddRequestSpecificInformation(Activity activity, IRequestContext re
             }
         }
 
-        if (AWSServiceHelper.IsDynamoDbService(service))
+        if (AWSServiceType.IsDynamoDbService(service))
         {
             activity.SetTag(SemanticConventions.AttributeDbSystem, AWSSemanticConventions.AttributeValueDynamoDb);
         }
+        else if (AWSServiceType.IsSqsService(service))
+        {
+            AWSMessagingUtils.Inject(new SqsRequestContextAdapter(requestContext), new PropagationContext(activity.Context, Baggage.Current));
+        }
+        else if (AWSServiceType.IsSnsService(service))
+        {
+            AWSMessagingUtils.Inject(new SnsRequestContextAdapter(requestContext), new PropagationContext(activity.Context, Baggage.Current));
+        }
     }
 
     private void AddStatusCodeToActivity(Activity activity, int status_code)

src/OpenTelemetry.Contrib.Instrumentation.AWS/Implementation/IRequestContextAdapter.cs

@@ -0,0 +1,31 @@
+// <copyright file="IRequestContextAdapter.cs" company="OpenTelemetry Authors">
+// 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.
+// </copyright>
+
+namespace OpenTelemetry.Contrib.Instrumentation.AWS.Implementation;
+internal interface IRequestContextAdapter
+{
+    bool HasMessageBody { get; }
+
+    bool HasOriginalRequest { get; }
+
+    int AttributesCount { get; }
+
+    bool ContainsAttribute(string name);
+
+    void AddAttribute(string name, string value, int nextAttributeIndex);
+
+    void AddAttributeToOriginalRequest(string name, string value);
+}

src/OpenTelemetry.Contrib.Instrumentation.AWS/Implementation/SnsRequestContextAdapter.cs

@@ -0,0 +1,59 @@
+// <copyright file="SnsRequestContextAdapter.cs" company="OpenTelemetry Authors">
+// 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.
+// </copyright>
+
+using Amazon.Runtime;
+using Amazon.Runtime.Internal;
+using Amazon.SimpleNotificationService.Model;
+
+namespace OpenTelemetry.Contrib.Instrumentation.AWS.Implementation;
+internal class SnsRequestContextAdapter : IRequestContextAdapter
+{
+    private readonly ParameterCollection parameters;
+    private readonly PublishRequest originalRequest;
+
+    public SnsRequestContextAdapter(IRequestContext context)
+    {
+        this.parameters = context.Request?.ParameterCollection;
+        this.originalRequest = context.OriginalRequest as PublishRequest;
+    }
+
+    public bool HasMessageBody =>
+        this.parameters?.ContainsKey("Message") ?? false;
+
+    public bool HasOriginalRequest => this.originalRequest != null;
+
+    public int AttributesCount =>
+        this.originalRequest?.MessageAttributes.Count ?? 0;
+
+    public void AddAttribute(string name, string value, int nextAttributeIndex)
+    {
+        if (this.parameters == null)
+        {
+            return;
+        }
+
+        var attributePrefix = "MessageAttributes.entry." + nextAttributeIndex;
+        this.parameters.Add(attributePrefix + ".Name", name);
+        this.parameters.Add(attributePrefix + ".Value.DataType", "String");
+        this.parameters.Add(attributePrefix + ".Value.StringValue", value);
+    }
+
+    public void AddAttributeToOriginalRequest(string name, string value) =>
+        this.originalRequest?.MessageAttributes.Add(name, new MessageAttributeValue { DataType = "String", StringValue = value });
+
+    public bool ContainsAttribute(string name)
+        => this.originalRequest?.MessageAttributes.ContainsKey(name) ?? false;
+}

src/OpenTelemetry.Contrib.Instrumentation.AWS/Implementation/SqsRequestContextAdapter.cs

@@ -0,0 +1,59 @@
+// <copyright file="SqsRequestContextAdapter.cs" company="OpenTelemetry Authors">
+// 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.
+// </copyright>
+
+using Amazon.Runtime;
+using Amazon.Runtime.Internal;
+using Amazon.SQS.Model;
+
+namespace OpenTelemetry.Contrib.Instrumentation.AWS.Implementation;
+internal class SqsRequestContextAdapter : IRequestContextAdapter
+{
+    private readonly ParameterCollection parameters;
+    private readonly SendMessageRequest originalRequest;
+
+    public SqsRequestContextAdapter(IRequestContext context)
+    {
+        this.parameters = context.Request?.ParameterCollection;
+        this.originalRequest = context.OriginalRequest as SendMessageRequest;
+    }
+
+    public bool HasMessageBody =>
+        this.parameters?.ContainsKey("MessageBody") ?? false;
+
+    public bool HasOriginalRequest => this.originalRequest != null;
+
+    public int AttributesCount =>
+        this.originalRequest?.MessageAttributes.Count ?? 0;
+
+    public void AddAttribute(string name, string value, int nextAttributeIndex)
+    {
+        if (this.parameters == null)
+        {
+            return;
+        }
+
+        var attributePrefix = "MessageAttribute." + nextAttributeIndex;
+        this.parameters.Add(attributePrefix + ".Name", name);
+        this.parameters.Add(attributePrefix + ".Value.DataType", "String");
+        this.parameters.Add(attributePrefix + ".Value.StringValue", value);
+    }
+
+    public void AddAttributeToOriginalRequest(string name, string value) =>
+        this.originalRequest?.MessageAttributes.Add(name, new MessageAttributeValue { DataType = "String", StringValue = value });
+
+    public bool ContainsAttribute(string name) =>
+        this.originalRequest?.MessageAttributes.ContainsKey(name) ?? false;
+}

src/OpenTelemetry.Contrib.Instrumentation.AWS/OpenTelemetry.Contrib.Instrumentation.AWS.csproj

@@ -8,12 +8,14 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="AWSSDK.SimpleNotificationService" Version="3.7.101.15" />
+    <PackageReference Include="AWSSDK.SQS" Version="3.7.100.79" />
     <PackageReference Include="OpenTelemetry.Contrib.Extensions.AWSXRay" Version="1.0.1" />
-    <PackageReference Include="AWSSDK.Core" Version="3.5.1.24" />
   </ItemGroup>
 
   <ItemGroup>
     <Compile Include="$(RepoRoot)\src\OpenTelemetry.Internal\Guard.cs" Link="Includes\Guard.cs" />
-    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\SemanticConventions.cs" Link="Includes\SemanticConventions.cs"/>
+    <Compile Include="$(RepoRoot)\src\OpenTelemetry.Contrib.Shared\Api\SemanticConventions.cs" Link="Includes\SemanticConventions.cs" />
   </ItemGroup>
+
 </Project>

src/OpenTelemetry.Instrumentation.AWSLambda/Implementation/AWSLambdaUtils.cs

@@ -20,6 +20,8 @@
 using System.Linq;
 using Amazon.Lambda.APIGatewayEvents;
 using Amazon.Lambda.Core;
+using Amazon.Lambda.SNSEvents;
+using Amazon.Lambda.SQSEvents;
 using OpenTelemetry.Context.Propagation;
 using OpenTelemetry.Contrib.Extensions.AWSXRay.Trace;
 
@@ -74,6 +76,18 @@ internal static ActivityContext ExtractParentContext<TInput>(TInput input)
             case APIGatewayHttpApiV2ProxyRequest apiGatewayHttpApiV2ProxyRequest:
                 propagationContext = Propagators.DefaultTextMapPropagator.Extract(default, apiGatewayHttpApiV2ProxyRequest, GetHeaderValues);
                 break;
+            case SQSEvent sqsEvent:
+                propagationContext = AWSMessagingUtils.ExtractParentContext(sqsEvent);
+                break;
+            case SQSEvent.SQSMessage sqsMessage:
+                propagationContext = AWSMessagingUtils.ExtractParentContext(sqsMessage);
+                break;
+            case SNSEvent snsEvent:
+                propagationContext = AWSMessagingUtils.ExtractParentContext(snsEvent);
+                break;
+            case SNSEvent.SNSRecord snsRecord:
+                propagationContext = AWSMessagingUtils.ExtractParentContext(snsRecord);
+                break;
         }
 
         return propagationContext.ActivityContext;