[Instrumentation.MassTransit] Masstransit enricher

src/OpenTelemetry.Instrumentation.MassTransit/CHANGELOG.md

@@ -7,6 +7,8 @@
 * Remove .NET Framework 4.6.1. The minimum .NET Framework version supported is
   .NET Framework 4.6.2 (by .NET Standard 2.0).
   ([#637](https://github.com/open-telemetry/opentelemetry-dotnet/issues/637))
+* Added enricher for Masstransit activity
+  ([#696](https://github.com/open-telemetry/opentelemetry-dotnet-contrib/pull/696))
 
 ## 1.0.0-beta.3
 

src/OpenTelemetry.Instrumentation.MassTransit/Implementation/MassTransitDiagnosticListener.cs

@@ -61,6 +61,15 @@ public override void OnStartActivity(Activity activity, object payload)
 
             ActivityInstrumentationHelper.SetActivitySourceProperty(activity, ActivitySource);
             ActivityInstrumentationHelper.SetKindProperty(activity, this.GetActivityKind(activity));
+
+            try
+            {
+                this.options.EnrichWithRequestPayload?.Invoke(activity, payload);
+            }
+            catch (Exception ex)
+            {
+                this.options.EnrichWithException(activity, ex);
+            }
         }
     }
 
@@ -71,10 +80,12 @@ public override void OnStopActivity(Activity activity, object payload)
             try
             {
                 this.TransformMassTransitTags(activity);
+
+                this.options.EnrichWithResponsePayload?.Invoke(activity, payload);
             }
             catch (Exception ex)
             {
-                MassTransitInstrumentationEventSource.Log.EnrichmentException(ex);
+                this.options.EnrichWithException(activity, ex);
             }
         }
     }

src/OpenTelemetry.Instrumentation.MassTransit/Implementation/MassTransitInstrumentationEventSource.cs

@@ -14,9 +14,7 @@
 // limitations under the License.
 // </copyright>
 
-using System;
 using System.Diagnostics.Tracing;
-using OpenTelemetry.Internal;
 
 namespace OpenTelemetry.Instrumentation.MassTransit.Implementation;
 
@@ -33,19 +31,4 @@ public void RequestIsFilteredOut(string eventName)
     {
         this.WriteEvent(1, eventName);
     }
-
-    [NonEvent]
-    public void EnrichmentException(Exception ex)
-    {
-        if (this.IsEnabled(EventLevel.Error, (EventKeywords)(-1)))
-        {
-            this.EnrichmentException(ex.ToInvariantString());
-        }
-    }
-
-    [Event(2, Message = "Enrich threw exception. Exception {0}.", Level = EventLevel.Error)]
-    public void EnrichmentException(string exception)
-    {
-        this.WriteEvent(2, exception);
-    }
 }

src/OpenTelemetry.Instrumentation.MassTransit/MassTransitInstrumentationOptions.cs

@@ -14,7 +14,10 @@
 // limitations under the License.
 // </copyright>
 
+using System;
 using System.Collections.Generic;
+using System.Diagnostics;
+using System.Net.Http;
 
 namespace OpenTelemetry.Instrumentation.MassTransit;
 
@@ -38,4 +41,32 @@ public class MassTransitInstrumentationOptions
     /// Gets or sets traced operations set.
     /// </summary>
     public HashSet<string> TracedOperations { get; set; } = new HashSet<string>(DefaultTracedOperations);
+
+    /// <summary>
+    /// Gets or sets an action to enrich an Activity.
+    /// </summary>
+    /// <remarks>
+    /// <para><see cref="Activity"/>: the activity being enriched.</para>
+    /// <para>object: the raw object from which additional information can be extracted to enrich the activity.
+    /// The type of this object depends on the event.</para>
+    /// </remarks>
+    public Action<Activity, object> EnrichWithRequestPayload { get; set; }
+
+    /// <summary>
+    /// Gets or sets an action to enrich an Activity with <see cref="HttpResponseMessage"/>.
+    /// </summary>
+    /// <remarks>
+    /// <para><see cref="Activity"/>: the activity being enriched.</para>
+    /// <para>object: the raw object from which additional information can be extracted to enrich the activity.</para>
+    /// </remarks>
+    public Action<Activity, object> EnrichWithResponsePayload { get; set; }
+
+    /// <summary>
+    /// Gets or sets an action to enrich an Activity with <see cref="Exception"/>.
+    /// </summary>
+    /// <remarks>
+    /// <para><see cref="Activity"/>: the activity being enriched.</para>
+    /// <para>object: the raw object from which additional information can be extracted to enrich the activity.</para>
+    /// </remarks>
+    public Action<Activity, object> EnrichWithException { get; set; }
 }

src/OpenTelemetry.Instrumentation.MassTransit/README.md

@@ -48,6 +48,27 @@ services.AddOpenTelemetrySdk(x =>
 });
 ```
 
+Instrumentation can be configured via options overload for
+`AddMassTransitInstrumentation` method:
+
+```csharp
+using var openTelemetry = Sdk.CreateTracerProviderBuilder()
+    .AddMassTransitInstrumentation(options =>
+    {
+        // Enable enriching an activity after it is created.
+        options.Enrich = (activity, eventName, rawObject) =>
+        {
+            if (eventName == "OnStartActivity")
+            {
+                var messagingOperation = activity.OperationName.Split('.').Last().ToLower();
+                if (messagingOperation != "send")
+                    activity.SetTag(SemanticConventions.AttributeMessagingOperation, messagingOperation);
+            }
+        };
+    })
+    .Build();
+```
+
 ## Filter traced operations
 
 For example you can trace only consume and handle operations using this snippet:

test/OpenTelemetry.Instrumentation.MassTransit.Tests/MassTransitInstrumentationTests.cs

@@ -14,6 +14,7 @@
 // limitations under the License.
 // </copyright>
 
+using System;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
@@ -226,11 +227,26 @@ public async Task ShouldMapMassTransitTagsForHandleMessageToOpenTelemetrySpecifi
 
     [Theory]
     [InlineData(OperationName.Consumer.Consume)]
+    [InlineData(OperationName.Consumer.Consume, true)]
+    [InlineData(OperationName.Consumer.Consume, true, true)]
     [InlineData(OperationName.Consumer.Handle)]
+    [InlineData(OperationName.Consumer.Handle, true)]
+    [InlineData(OperationName.Consumer.Handle, true, true)]
     [InlineData(OperationName.Transport.Send)]
+    [InlineData(OperationName.Transport.Send, true)]
+    [InlineData(OperationName.Transport.Send, true, true)]
     [InlineData(OperationName.Transport.Receive)]
-    public async Task MassTransitInstrumentationTestOptions(string operationName)
+    [InlineData(OperationName.Transport.Receive, true)]
+    [InlineData(OperationName.Transport.Receive, true, true)]
+    public async Task MassTransitInstrumentationTestOptions(
+            string operationName,
+            bool enrich = false,
+            bool enrichmentException = false)
     {
+        bool enrichWithHttpRequestMessageCalled = false;
+        bool enrichWithHttpResponseMessageCalled = false;
+        bool enrichWithExceptionCalled = false;
+
         using Activity activity = new Activity("Parent");
         activity.SetParentId(
             ActivityTraceId.CreateRandom(),
@@ -240,10 +256,31 @@ public async Task MassTransitInstrumentationTestOptions(string operationName)
 
         var activityProcessor = new Mock<BaseProcessor<Activity>>();
         using (Sdk.CreateTracerProviderBuilder()
-                   .AddProcessor(activityProcessor.Object)
-                   .AddMassTransitInstrumentation(o =>
-                       o.TracedOperations = new HashSet<string>(new[] { operationName }))
-                   .Build())
+            .AddProcessor(activityProcessor.Object)
+            .AddMassTransitInstrumentation(o =>
+            {
+                o.TracedOperations = new HashSet<string>(new[] { operationName });
+
+                if (enrich)
+                {
+                    if (!enrichmentException)
+                    {
+                        o.EnrichWithRequestPayload = (activity, httpRequestMessage) => { enrichWithHttpRequestMessageCalled = true; };
+                    }
+                    else
+                    {
+                        o.EnrichWithRequestPayload = (activity, httpRequestMessage) =>
+                        {
+                            enrichWithHttpRequestMessageCalled = true;
+                            throw new Exception("Error while enriching activity");
+                        };
+                    }
+
+                    o.EnrichWithResponsePayload = (activity, httpResponseMessage) => { enrichWithHttpResponseMessageCalled = true; };
+                    o.EnrichWithException = (activity, obj) => { enrichWithExceptionCalled = true; };
+                }
+            })
+            .Build())
         {
             var harness = new InMemoryTestHarness();
             var consumerHarness = harness.Consumer<TestConsumer>();
@@ -271,6 +308,26 @@ await harness.InputQueueSendEndpoint.Send<TestMessage>(new
         var consumes = this.GetActivitiesFromInvocationsByOperationName(activityProcessor.Invocations, operationName);
 
         Assert.Single(consumes);
+
+        if (enrich)
+        {
+            Assert.True(enrichWithHttpRequestMessageCalled);
+            Assert.True(enrichWithHttpResponseMessageCalled);
+        }
+        else
+        {
+            Assert.False(enrichWithHttpRequestMessageCalled);
+            Assert.False(enrichWithHttpResponseMessageCalled);
+        }
+
+        if (enrichmentException)
+        {
+            Assert.True(enrichWithExceptionCalled);
+        }
+        else
+        {
+            Assert.False(enrichWithExceptionCalled);
+        }
     }
 
     [Fact]