Add message related properties to transactions and spans (#1525)

Closes #1512

src/Elastic.Apm.Azure.ServiceBus/AzureMessagingServiceBusDiagnosticListener.cs

@@ -112,6 +112,9 @@ private void OnProcessStart(KeyValuePair<string, object> kv, string action)
 			var transaction = ApmAgent.Tracer.StartTransaction(transactionName, ApiConstants.TypeMessaging);
 			transaction.Context.Service = new Service(null, null) { Framework = _framework };
 
+			if (queueName != null)
+				transaction.Context.Message = new Message { Queue = new Queue { Name = queueName } };
+
 			// transaction creation will create an activity, so use this as the key.
 			var activityId = Activity.Current.Id;
 
@@ -158,11 +161,15 @@ private void OnReceiveStart(KeyValuePair<string, object> kv, string action)
 			{
 				var transaction = ApmAgent.Tracer.StartTransaction(transactionName, ApiConstants.TypeMessaging);
 				transaction.Context.Service = new Service(null, null) { Framework = _framework };
+				if (queueName != null)
+					transaction.Context.Message = new Message { Queue = new Queue { Name = queueName } };
 				segment = transaction;
 			}
 			else
 			{
 				var span = ApmAgent.GetCurrentExecutionSegment().StartSpan(transactionName, ApiConstants.TypeMessaging, ServiceBus.SubType, action);
+				if (queueName != null)
+					span.Context.Message = new Message { Queue = new Queue { Name = queueName } };
 				segment = span;
 			}
 
@@ -249,6 +256,9 @@ private void OnSendStart(KeyValuePair<string, object> kv, string action)
 				}
 			};
 
+			if (queueName != null)
+				span.Context.Message = new Message { Queue = new Queue { Name = queueName } };
+
 			if (!_processingSegments.TryAdd(activity.Id, span))
 			{
 				Logger.Trace()?.Log(

src/Elastic.Apm.Azure.ServiceBus/MicrosoftAzureServiceBusDiagnosticListener.cs

@@ -116,6 +116,9 @@ private void OnProcessStart(KeyValuePair<string, object> kv, string action, Prop
 			var transaction = ApmAgent.Tracer.StartTransaction(transactionName, ApiConstants.TypeMessaging);
 			transaction.Context.Service = new Service(null, null) { Framework = _framework };
 
+			if (queueName != null)
+				transaction.Context.Message = new Message { Queue = new Queue { Name = queueName } };
+
 			// transaction creation will create an activity, so use this as the key.
 			var activityId = Activity.Current.Id;
 
@@ -150,11 +153,15 @@ private void OnReceiveStart(KeyValuePair<string, object> kv, string action, Prop
 			{
 				var transaction = ApmAgent.Tracer.StartTransaction(transactionName, ApiConstants.TypeMessaging);
 				transaction.Context.Service = new Service(null, null) { Framework = _framework };
+				if (queueName != null)
+					transaction.Context.Message = new Message { Queue = new Queue { Name = queueName } };
 				segment = transaction;
 			}
 			else
 			{
 				var span = ApmAgent.GetCurrentExecutionSegment().StartSpan(transactionName, ApiConstants.TypeMessaging, ServiceBus.SubType, action);
+				if (queueName != null)
+					span.Context.Message = new Message { Queue = new Queue { Name = queueName } };
 				segment = span;
 			}
 
@@ -229,6 +236,9 @@ private void OnSendStart(KeyValuePair<string, object> kv, string action, Propert
 				}
 			};
 
+			if (queueName != null)
+				span.Context.Message = new Message { Queue = new Queue { Name = queueName } };
+
 			if (!_processingSegments.TryAdd(activity.Id, span))
 			{
 				Logger.Trace()?.Log(

src/Elastic.Apm/Api/Context.cs

@@ -27,6 +27,11 @@ public class Context
 		[JsonProperty("tags")]
 		public Dictionary<string, string> Labels => InternalLabels.Value;
 
+		/// <summary>
+		/// Holds details related to message receiving and publishing if the captured event integrates with a messaging system
+		/// </summary>
+		public Message Message { get; set; }
+
 		/// <summary>
 		/// If a log record was generated as a result of a http request, the http interface can be used to collect this
 		/// information.

src/Elastic.Apm/Api/Message.cs

@@ -0,0 +1,62 @@
+// Licensed to Elasticsearch B.V under
+// one or more agreements.
+// Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information
+
+using System.Collections.Generic;
+using Elastic.Apm.Api.Constraints;
+
+namespace Elastic.Apm.Api
+{
+	/// <summary>
+	/// Holds details related to message receiving and publishing if the captured event integrates with a messaging system
+	/// </summary>
+	public class Message
+	{
+		/// <summary>
+		/// Body of the received message
+		/// </summary>
+		public string Body { get; set; }
+
+		/// <summary>
+		/// Headers received with the message
+		/// </summary>
+		public Dictionary<string, string> Headers { get; set; }
+
+		/// <inheritdoc cref="Api.Age"/>
+		public Age Age { get; set; }
+
+		/// <see cref="Api.Queue"/>
+		public Queue Queue { get; set; }
+
+		/// <summary>
+		/// optional routing key of the received message as set on the queuing system, such as in RabbitMQ.
+		/// </summary>
+		public string RoutingKey { get; set; }
+	}
+
+	/// <summary>
+	/// Age of the message. If the monitored messaging framework provides a timestamp for the message, agents may use it.
+	/// Otherwise, the sending agent can add a timestamp in milliseconds since the Unix epoch to the message's metadata to be retrieved by the
+	/// receiving agent. If a timestamp is not available, agents should omit this field.
+	/// </summary>
+	public class Age
+	{
+		/// <summary>
+		/// Age of the message in milliseconds.
+		/// </summary>
+		public long Ms { get; set; }
+	}
+
+	/// <summary>
+	/// Information about the message queue where the message is received.
+	/// </summary>
+	public class Queue
+	{
+		/// <summary>
+		/// Name of the message queue where the message is received
+		/// </summary>
+		[MaxLength]
+		public string Name { get; set; }
+	}
+}

src/Elastic.Apm/Api/Request.cs

@@ -7,7 +7,6 @@
 using System.Linq;
 using Elastic.Apm.Helpers;
 using Elastic.Apm.Libraries.Newtonsoft.Json;
-using Elastic.Apm.Libraries.Newtonsoft.Json.Serialization;
 
 namespace Elastic.Apm.Api
 {

src/Elastic.Apm/Api/SpanContext.cs

@@ -17,6 +17,7 @@ public class SpanContext
 		public Database Db { get; set; }
 		public Destination Destination { get; set; }
 		public Http Http { get; set; }
+		public Message Message { get; set; }
 
 		/// <summary>
 		/// <seealso cref="ShouldSerializeLabels" />

src/Elastic.Apm/Filters/ErrorContextSanitizerFilter.cs

@@ -18,12 +18,24 @@ internal class ErrorContextSanitizerFilter
 	{
 		public IError Filter(IError error)
 		{
-			if (error is Error realError && realError.Context?.Request?.Headers != null && realError.Configuration != null)
+			if (error is Error realError && realError.Configuration != null)
 			{
-				foreach (var key in realError.Context.Request.Headers.Keys.ToList())
+				if (realError.Context?.Request?.Headers != null)
 				{
-					if (WildcardMatcher.IsAnyMatch(realError.Configuration.SanitizeFieldNames, key))
-						realError.Context.Request.Headers[key] = Consts.Redacted;
+					foreach (var key in realError.Context.Request.Headers.Keys.ToList())
+					{
+						if (WildcardMatcher.IsAnyMatch(realError.Configuration.SanitizeFieldNames, key))
+							realError.Context.Request.Headers[key] = Consts.Redacted;
+					}
+				}
+
+				if (realError.Context?.Message?.Headers != null)
+				{
+					foreach (var key in realError.Context.Message.Headers.Keys.ToList())
+					{
+						if (WildcardMatcher.IsAnyMatch(realError.Configuration.SanitizeFieldNames, key))
+							realError.Context.Message.Headers[key] = Consts.Redacted;
+					}
 				}
 			}
 

src/Elastic.Apm/Filters/HeaderDictionarySanitizerFilter.cs

@@ -20,12 +20,24 @@ public ITransaction Filter(ITransaction transaction)
 		{
 			if (transaction is Transaction realTransaction)
 			{
-				if (realTransaction.IsContextCreated && realTransaction.Context.Request?.Headers != null)
+				if (realTransaction.IsContextCreated)
 				{
-					foreach (var key in realTransaction.Context?.Request?.Headers?.Keys.ToList())
+					if (realTransaction.Context.Request?.Headers != null)
 					{
-						if (WildcardMatcher.IsAnyMatch(realTransaction.Configuration.SanitizeFieldNames, key))
-							realTransaction.Context.Request.Headers[key] = Consts.Redacted;
+						foreach (var key in realTransaction.Context.Request.Headers.Keys.ToList())
+						{
+							if (WildcardMatcher.IsAnyMatch(realTransaction.Configuration.SanitizeFieldNames, key))
+								realTransaction.Context.Request.Headers[key] = Consts.Redacted;
+						}
+					}
+
+					if (realTransaction.Context.Message?.Headers != null)
+					{
+						foreach (var key in realTransaction.Context.Message.Headers.Keys.ToList())
+						{
+							if (WildcardMatcher.IsAnyMatch(realTransaction.Configuration.SanitizeFieldNames, key))
+								realTransaction.Context.Message.Headers[key] = Consts.Redacted;
+						}
 					}
 				}
 			}

test/Elastic.Apm.Azure.ServiceBus.Tests/AzureMessagingServiceBusDiagnosticListenerTests.cs

@@ -64,6 +64,10 @@ await _agent.Tracer.CaptureTransaction("Send AzureServiceBus Message", "message"
 			destination.Service.Name.Should().Be(ServiceBus.SubType);
 			destination.Service.Resource.Should().Be($"{ServiceBus.SubType}/{scope.QueueName}");
 			destination.Service.Type.Should().Be(ApiConstants.TypeMessaging);
+
+			span.Context.Message.Should().NotBeNull();
+			span.Context.Message.Queue.Should().NotBeNull();
+			span.Context.Message.Queue.Name.Should().Be(scope.QueueName);
 		}
 
 		[AzureCredentialsFact]
@@ -93,6 +97,10 @@ await _agent.Tracer.CaptureTransaction("Send AzureServiceBus Message", "message"
 			destination.Service.Name.Should().Be(ServiceBus.SubType);
 			destination.Service.Resource.Should().Be($"{ServiceBus.SubType}/{scope.TopicName}");
 			destination.Service.Type.Should().Be(ApiConstants.TypeMessaging);
+
+			span.Context.Message.Should().NotBeNull();
+			span.Context.Message.Queue.Should().NotBeNull();
+			span.Context.Message.Queue.Name.Should().Be(scope.TopicName);
 		}
 
 		[AzureCredentialsFact]
@@ -124,6 +132,10 @@ await sender.ScheduleMessageAsync(
 			destination.Service.Name.Should().Be(ServiceBus.SubType);
 			destination.Service.Resource.Should().Be($"{ServiceBus.SubType}/{scope.QueueName}");
 			destination.Service.Type.Should().Be(ApiConstants.TypeMessaging);
+
+			span.Context.Message.Should().NotBeNull();
+			span.Context.Message.Queue.Should().NotBeNull();
+			span.Context.Message.Queue.Name.Should().Be(scope.QueueName);
 		}
 
 		[AzureCredentialsFact]
@@ -155,6 +167,10 @@ await sender.ScheduleMessageAsync(
 			destination.Service.Name.Should().Be(ServiceBus.SubType);
 			destination.Service.Resource.Should().Be($"{ServiceBus.SubType}/{scope.TopicName}");
 			destination.Service.Type.Should().Be(ApiConstants.TypeMessaging);
+
+			span.Context.Message.Should().NotBeNull();
+			span.Context.Message.Queue.Should().NotBeNull();
+			span.Context.Message.Queue.Name.Should().Be(scope.TopicName);
 		}
 
 		[AzureCredentialsFact]
@@ -182,6 +198,10 @@ await _agent.Tracer.CaptureTransaction("Receive messages", ApiConstants.TypeMess
 			span.Name.Should().Be($"{ServiceBus.SegmentName} RECEIVE from {scope.QueueName}");
 			span.Type.Should().Be(ApiConstants.TypeMessaging);
 			span.Subtype.Should().Be(ServiceBus.SubType);
+
+			span.Context.Message.Should().NotBeNull();
+			span.Context.Message.Queue.Should().NotBeNull();
+			span.Context.Message.Queue.Name.Should().Be(scope.QueueName);
 		}
 
 		[AzureCredentialsFact]
@@ -204,6 +224,10 @@ await sender.SendMessageAsync(
 
 			transaction.Name.Should().Be($"{ServiceBus.SegmentName} RECEIVE from {scope.QueueName}");
 			transaction.Type.Should().Be(ApiConstants.TypeMessaging);
+
+			transaction.Context.Message.Should().NotBeNull();
+			transaction.Context.Message.Queue.Should().NotBeNull();
+			transaction.Context.Message.Queue.Name.Should().Be(scope.QueueName);
 		}
 
 		[AzureCredentialsFact]
@@ -224,9 +248,13 @@ await sender.SendMessageAsync(
 
 			_sender.Transactions.Should().HaveCount(1);
 			var transaction = _sender.FirstTransaction;
-
-			transaction.Name.Should().Be($"{ServiceBus.SegmentName} RECEIVE from {scope.TopicName}/Subscriptions/{scope.SubscriptionName}");
+			var subscription = $"{scope.TopicName}/Subscriptions/{scope.SubscriptionName}";
+			transaction.Name.Should().Be($"{ServiceBus.SegmentName} RECEIVE from {subscription}");
 			transaction.Type.Should().Be(ApiConstants.TypeMessaging);
+
+			transaction.Context.Message.Should().NotBeNull();
+			transaction.Context.Message.Queue.Should().NotBeNull();
+			transaction.Context.Message.Queue.Name.Should().Be(subscription);
 		}
 
 		[AzureCredentialsFact]
@@ -242,8 +270,6 @@ await sender.SendMessageAsync(
 
 			var message = await receiver.ReceiveMessageAsync(TimeSpan.FromSeconds(30)).ConfigureAwait(false);
 			await receiver.DeferMessageAsync(message).ConfigureAwait(false);
-
-
 			await receiver.ReceiveDeferredMessageAsync(message.SequenceNumber).ConfigureAwait(false);
 
 			if (!_sender.WaitForTransactions(TimeSpan.FromMinutes(2), count: 2))
@@ -255,6 +281,10 @@ await sender.SendMessageAsync(
 			transaction.Name.Should().Be($"{ServiceBus.SegmentName} RECEIVE from {scope.QueueName}");
 			transaction.Type.Should().Be(ApiConstants.TypeMessaging);
 
+			transaction.Context.Message.Should().NotBeNull();
+			transaction.Context.Message.Queue.Should().NotBeNull();
+			transaction.Context.Message.Queue.Name.Should().Be(scope.QueueName);
+
 			var secondTransaction = _sender.Transactions[1];
 			secondTransaction.Name.Should().Be($"{ServiceBus.SegmentName} RECEIVEDEFERRED from {scope.QueueName}");
 			secondTransaction.Type.Should().Be(ApiConstants.TypeMessaging);
@@ -282,9 +312,14 @@ await sender.SendMessageAsync(
 			_sender.Transactions.Should().HaveCount(2);
 
 			var transaction = _sender.FirstTransaction;
-			transaction.Name.Should().Be($"{ServiceBus.SegmentName} RECEIVE from {scope.TopicName}/Subscriptions/{scope.SubscriptionName}");
+			var subscription = $"{scope.TopicName}/Subscriptions/{scope.SubscriptionName}";
+			transaction.Name.Should().Be($"{ServiceBus.SegmentName} RECEIVE from {subscription}");
 			transaction.Type.Should().Be(ApiConstants.TypeMessaging);
 
+			transaction.Context.Message.Should().NotBeNull();
+			transaction.Context.Message.Queue.Should().NotBeNull();
+			transaction.Context.Message.Queue.Name.Should().Be(subscription);
+
 			var secondTransaction = _sender.Transactions[1];
 			secondTransaction.Name.Should().Be($"{ServiceBus.SegmentName} RECEIVEDEFERRED from {scope.TopicName}/Subscriptions/{scope.SubscriptionName}");
 			secondTransaction.Type.Should().Be(ApiConstants.TypeMessaging);
@@ -353,6 +388,10 @@ public async Task Capture_Transaction_When_ProcessMessage_From_Queue()
 
 			foreach (var transaction in processTransactions)
 			{
+				transaction.Context.Message.Should().NotBeNull();
+				transaction.Context.Message.Queue.Should().NotBeNull();
+				transaction.Context.Message.Queue.Name.Should().Be(scope.QueueName);
+
 				var spans = _sender.Spans.Where(s => s.TransactionId == transaction.Id).ToList();
 				spans.Should().HaveCount(1);
 			}
@@ -409,6 +448,10 @@ public async Task Capture_Transaction_When_ProcessSessionMessage_From_Queue()
 
 			foreach (var transaction in processTransactions)
 			{
+				transaction.Context.Message.Should().NotBeNull();
+				transaction.Context.Message.Queue.Should().NotBeNull();
+				transaction.Context.Message.Queue.Name.Should().Be(scope.QueueName);
+
 				var spans = _sender.Spans.Where(s => s.TransactionId == transaction.Id).ToList();
 				spans.Should().HaveCount(1);
 			}