#1034: AWS SQS and SNS support

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

@@ -0,0 +1,35 @@
+// <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
+{
+    internal static void Inject(IRequestContextAdapter request, PropagationContext propagationContext)
+    {
+        if (!request.CanInject)
+        {
+            return;
+        }
+
+        var carrier = new Dictionary<string, string>();
+        Propagators.DefaultTextMapPropagator.Inject(propagationContext, carrier, (c, k, v) => c[k] = v);
+        request.AddAttributes(carrier);
+    }
+}

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,25 @@
+// <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>
+
+using System.Collections.Generic;
+
+namespace OpenTelemetry.Contrib.Instrumentation.AWS.Implementation;
+internal interface IRequestContextAdapter
+{
+    bool CanInject { get; }
+
+    void AddAttributes(IReadOnlyDictionary<string, string> attributes);
+}

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

@@ -0,0 +1,89 @@
+// <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 System.Collections.Generic;
+using System.Linq;
+using Amazon.Runtime;
+using Amazon.Runtime.Internal;
+using Amazon.SimpleNotificationService.Model;
+
+namespace OpenTelemetry.Contrib.Instrumentation.AWS.Implementation;
+internal class SnsRequestContextAdapter : IRequestContextAdapter
+{
+    // 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;
+
+    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 CanInject => this.originalRequest?.MessageAttributes != null && this.parameters != null;
+
+    public void AddAttributes(IReadOnlyDictionary<string, string> attributes)
+    {
+        if (!this.CanInject)
+        {
+            return;
+        }
+
+        if (attributes.Keys.Any(k => this.ContainsAttribute(k)))
+        {
+            // If at least one attribute is already present in the request then we skip the injection.
+            return;
+        }
+
+        int attributesCount = this.originalRequest?.MessageAttributes.Count ?? 0;
+        if (attributes.Count + attributesCount > MaxMessageAttributes)
+        {
+            // TODO: add logging (event source).
+            return;
+        }
+
+        int nextAttributeIndex = attributesCount + 1;
+        foreach (var param in attributes)
+        {
+            this.AddAttribute(param.Key, param.Value, nextAttributeIndex);
+            nextAttributeIndex++;
+        }
+    }
+
+    private void AddAttribute(string name, string value, int nextAttributeIndex)
+    {
+        if (!this.CanInject)
+        {
+            return;
+        }
+
+        var prefix = "MessageAttributes.entry." + nextAttributeIndex;
+        this.parameters?.Add(prefix + ".Name", name);
+        this.parameters?.Add(prefix + ".Value.DataType", "String");
+        this.parameters?.Add(prefix + ".Value.StringValue", value);
+
+        // Add injected attributes to the original request as well.
+        // This dictionary must be in sync with parameters collection to pass through the MD5 hash matching check.
+        this.originalRequest?.MessageAttributes.Add(name, new MessageAttributeValue { DataType = "String", StringValue = value });
+    }
+
+    private bool ContainsAttribute(string name)
+        => this.originalRequest?.MessageAttributes.ContainsKey(name) ?? false;
+}

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

@@ -0,0 +1,89 @@
+// <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 System.Collections.Generic;
+using System.Linq;
+using Amazon.Runtime;
+using Amazon.Runtime.Internal;
+using Amazon.SQS.Model;
+
+namespace OpenTelemetry.Contrib.Instrumentation.AWS.Implementation;
+internal class SqsRequestContextAdapter : IRequestContextAdapter
+{
+    // 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;
+
+    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 CanInject => this.originalRequest?.MessageAttributes != null && this.parameters != null;
+
+    public void AddAttributes(IReadOnlyDictionary<string, string> attributes)
+    {
+        if (!this.CanInject)
+        {
+            return;
+        }
+
+        if (attributes.Keys.Any(k => this.ContainsAttribute(k)))
+        {
+            // If at least one attribute is already present in the request then we skip the injection.
+            return;
+        }
+
+        int attributesCount = this.originalRequest?.MessageAttributes.Count ?? 0;
+        if (attributes.Count + attributesCount > MaxMessageAttributes)
+        {
+            // TODO: add logging (event source).
+            return;
+        }
+
+        int nextAttributeIndex = attributesCount + 1;
+        foreach (var param in attributes)
+        {
+            this.AddAttribute(param.Key, param.Value, nextAttributeIndex);
+            nextAttributeIndex++;
+        }
+    }
+
+    private void AddAttribute(string name, string value, int attributeIndex)
+    {
+        if (!this.CanInject)
+        {
+            return;
+        }
+
+        var prefix = "MessageAttribute." + attributeIndex;
+        this.parameters?.Add(prefix + ".Name", name);
+        this.parameters?.Add(prefix + ".Value.DataType", "String");
+        this.parameters?.Add(prefix + ".Value.StringValue", value);
+
+        // Add injected attributes to the original request as well.
+        // This dictionary must be in sync with parameters collection to pass through the MD5 hash matching check.
+        this.originalRequest?.MessageAttributes.Add(name, new MessageAttributeValue { DataType = "String", StringValue = value });
+    }
+
+    private bool ContainsAttribute(string name) =>
+        this.originalRequest?.MessageAttributes.ContainsKey(name) ?? false;
+}

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

@@ -9,12 +9,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;