Emit pipeline and message operation events for diagnostics purposes

danielmarbach · danielmarbach · commit 3f550121474c · 2019-10-17T09:31:11.000+02:00
diff --git a/src/NServiceBus.Core/NServiceBus.Core.csproj b/src/NServiceBus.Core/NServiceBus.Core.csproj
@@ -5,6 +5,7 @@
     <RootNamespace>NServiceBus</RootNamespace>
     <SignAssembly>true</SignAssembly>
     <AssemblyOriginatorKeyFile>..\NServiceBus.snk</AssemblyOriginatorKeyFile>
+    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(TargetFramework)' == 'net452'">
diff --git a/src/NServiceBus.Core/Pipeline/MainPipelineExecutor.cs b/src/NServiceBus.Core/Pipeline/MainPipelineExecutor.cs
@@ -25,8 +25,12 @@ public async Task Invoke(MessageContext messageContext)
                 var rootContext = pipelineComponent.CreateRootContext(childBuilder, messageContext.Extensions);
                 var transportReceiveContext = new TransportReceiveContext(message, messageContext.TransportTransaction, messageContext.ReceiveCancellationTokenSource, rootContext);
 
+                var isFaulted = false;
+                var pipelineEventSource = PipelineEventSource.Log;
                 try
                 {
+                    pipelineEventSource.MainStart(messageContext.MessageId);
+
                     await transportReceiveContext.InvokePipeline<ITransportReceiveContext>().ConfigureAwait(false);
                 }
                 catch (Exception e)
@@ -39,6 +43,10 @@ public async Task Invoke(MessageContext messageContext)
 
                     throw;
                 }
+                finally
+                {
+                    pipelineEventSource.MainStop(messageContext.MessageId, isFaulted);
+                }
 
                 await transportReceiveContext.RaiseNotification(new ReceivePipelineCompleted(message, pipelineStartedAt, DateTime.UtcNow)).ConfigureAwait(false);
             }
diff --git a/src/NServiceBus.Core/Pipeline/Pipeline.cs b/src/NServiceBus.Core/Pipeline/Pipeline.cs
@@ -29,11 +29,32 @@ public Pipeline(IBuilder builder, PipelineModifications pipelineModifications)
 
         public Task Invoke(TContext context)
         {
-            return pipeline(context);
+            return !PipelineEventSource.Log.IsEnabled() ? pipeline.Invoke(context) : InvokePipelineAndEmitEvents(context);
+        }
+
+        async Task InvokePipelineAndEmitEvents(TContext context)
+        {
+            var isFaulted = false;
+            var pipelineEventSource = PipelineEventSource.Log;
+            pipelineEventSource.InvokeStart(pipelineName);
+            try
+            {
+                await pipeline.Invoke(context).ConfigureAwait(false);
+            }
+            catch
+            {
+                isFaulted = true;
+                throw;
+            }
+            finally
+            {
+                pipelineEventSource.InvokeStop(pipelineName, isFaulted);
+            }
         }
 
         // ReSharper disable once PrivateFieldCanBeConvertedToLocalVariable
         IBehavior[] behaviors;
         Func<TContext, Task> pipeline;
+        string pipelineName = typeof(TContext).FullName;
     }
 }
diff --git a/src/NServiceBus.Core/Pipeline/PipelineEventSource.cs b/src/NServiceBus.Core/Pipeline/PipelineEventSource.cs
@@ -0,0 +1,118 @@
+namespace NServiceBus
+{
+    using System;
+    using System.Diagnostics.CodeAnalysis;
+    using System.Diagnostics.Tracing;
+    using System.Runtime.CompilerServices;
+
+    /// <summary>
+    /// Do not rename the methods or parameters on this class nor change the event declarations because it directly affects
+    /// the events that are generated.
+    /// </summary>
+    [EventSource(Name = EventSourceName)]
+    [SuppressMessage("ReSharper", "ParameterHidesMember")]
+    sealed class PipelineEventSource : EventSource
+    {
+        PipelineEventSource()
+        {
+        }
+
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        [Event(MainPipelineStartEventId, Message = "Main Pipeline for MessageId '{0}' started.", Level = EventLevel.Informational)]
+        public void MainStart(string MessageId) // used within already existing state machine. Enabled check done as part of method implementation
+        {
+            if (IsEnabled())
+            {
+                WriteEvent(MainPipelineStartEventId, MessageId);
+            }
+        }
+
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        [Event(MainPipelineStopEventId, Message = "Main Pipeline for MessageId '{0}' stopped.", Level = EventLevel.Informational)]
+        public unsafe void MainStop(string MessageId, bool IsFaulted) // used within already existing state machine. Enabled check done as part of method implementation
+        {
+            if (IsEnabled())
+            {
+                fixed(char* messageIdPtr = MessageId)
+                {
+                    var eventPayload = stackalloc EventData[2];
+
+                    eventPayload[0].Size = (MessageId.Length + 1) * 2;
+                    eventPayload[0].DataPointer = (IntPtr)messageIdPtr;
+
+                    eventPayload[1].Size = sizeof(bool);
+                    eventPayload[1].DataPointer = (IntPtr)(&IsFaulted);
+                    WriteEventCore(MainPipelineStopEventId, 2, eventPayload);
+                }
+            }
+        }
+
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        [Event(SatellitePipelineStartEventId, Message = "Satellite Pipeline '{0}' for MessageId '{1}' started.", Level = EventLevel.Informational)]
+        public void SatelliteStart(string Name, string MessageId) // used on on hot path where async state machine was optimized away. Enable check done as part of caller
+        {
+            WriteEvent(SatellitePipelineStartEventId, Name, MessageId);
+        }
+
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        [Event(SatellitePipelineStopEventId, Message = "Satellite Pipeline '{0}' for MessageId '{1}' stopped.", Level = EventLevel.Informational)]
+        public unsafe void SatelliteStop(string Name, string MessageId, bool IsFaulted) // used on on hot path where async state machine was optimized away. Enable check done as part of caller
+        {
+            fixed(char* namePtr = Name)
+            fixed(char* messageIdPtr = MessageId)
+            {
+                var eventPayload = stackalloc EventData[3];
+
+                eventPayload[0].Size = (Name.Length + 1) * 2;
+                eventPayload[0].DataPointer = (IntPtr)namePtr;
+
+                eventPayload[1].Size = (MessageId.Length + 1) * 2;
+                eventPayload[1].DataPointer = (IntPtr)messageIdPtr;
+
+                eventPayload[2].Size = sizeof(bool);
+                eventPayload[2].DataPointer = (IntPtr)(&IsFaulted);
+                WriteEventCore(SatellitePipelineStopEventId, 3, eventPayload);
+            }
+        }
+
+
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        [Event(PipelineStartEventId, Message = "Pipeline '{0}' started.'", Level = EventLevel.Verbose
+#if NETSTANDARD
+            , ActivityOptions = EventActivityOptions.Recursive // Pipelines are started within pipeline, to avoid auto-stop we need recursive on the platform it is available
+#endif
+        )]
+        public void InvokeStart(string Name) // used on on hot path where async state machine was optimized away. Enable check done as part of caller
+        {
+            WriteEvent(PipelineStartEventId, Name);
+        }
+
+
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        [Event(PipelineStopEventId, Message = "Pipeline '{0}' stopped.'", Level = EventLevel.Verbose)]
+        public unsafe void InvokeStop(string Name, bool IsFaulted) // used on on hot path where async state machine was optimized away. Enable check done as part of caller
+        {
+            fixed(char* namePtr = Name)
+            {
+                var eventPayload = stackalloc EventData[2];
+
+                eventPayload[0].Size = (Name.Length + 1) * 2;
+                eventPayload[0].DataPointer = (IntPtr)namePtr;
+
+                eventPayload[1].Size = sizeof(bool);
+                eventPayload[1].DataPointer = (IntPtr)(&IsFaulted);
+                WriteEventCore(PipelineStopEventId, 2, eventPayload);
+            }
+        }
+
+        const string EventSourceName = "NServiceBus.Pipeline";
+        const int MainPipelineStartEventId = 1;
+        const int MainPipelineStopEventId = 2;
+        const int PipelineStartEventId = 3;
+        const int PipelineStopEventId = 4;
+        const int SatellitePipelineStartEventId = 5;
+        const int SatellitePipelineStopEventId = 6;
+
+        internal static readonly PipelineEventSource Log = new PipelineEventSource();
+    }
+}
diff --git a/src/NServiceBus.Core/Pipeline/SatellitePipelineExecutor.cs b/src/NServiceBus.Core/Pipeline/SatellitePipelineExecutor.cs
@@ -16,9 +16,30 @@ public Task Invoke(MessageContext messageContext)
         {
             messageContext.Extensions.Set(messageContext.TransportTransaction);
 
-            return satelliteDefinition.OnMessage(builder, messageContext);
+            return !PipelineEventSource.Log.IsEnabled() ? satelliteDefinition.OnMessage(builder, messageContext) : InvokePipelineAndEmitEvents(messageContext);
         }
 
+        async Task InvokePipelineAndEmitEvents(MessageContext messageContext)
+        {
+            var isFaulted = false;
+            var pipelineEventSource = PipelineEventSource.Log;
+            try
+            {
+                pipelineEventSource.SatelliteStart(satelliteDefinition.Name, messageContext.MessageId);
+                await satelliteDefinition.OnMessage(builder, messageContext).ConfigureAwait(false);
+            }
+            catch
+            {
+                isFaulted = true;
+                throw;
+            }
+            finally
+            {
+                pipelineEventSource.SatelliteStop(satelliteDefinition.Name, messageContext.MessageId, isFaulted);
+            }
+        }
+
+
         SatelliteDefinition satelliteDefinition;
         IBuilder builder;
     }
diff --git a/src/NServiceBus.Core/Unicast/MessageOperations.cs b/src/NServiceBus.Core/Unicast/MessageOperations.cs
@@ -39,7 +39,26 @@ static Task Publish(IBehaviorContext context, Type messageType, object message,
                 options.Context,
                 context);
 
-            return publishContext.InvokePipeline<IOutgoingPublishContext>();
+            return !MessageOperationsEventSource.Log.IsEnabled() ? publishContext.InvokePipeline<IOutgoingPublishContext>() : InvokePipelineAndEmitEvents(publishContext);
+        }
+
+        static async Task InvokePipelineAndEmitEvents(IOutgoingPublishContext outgoingContext)
+        {
+            var isFaulted = false;
+            var messageOperationsEventSource = MessageOperationsEventSource.Log;
+            try
+            {
+                messageOperationsEventSource.PublishStart(outgoingContext.MessageId);
+                await outgoingContext.InvokePipeline().ConfigureAwait(false);
+            }
+            catch
+            {
+                isFaulted = true;
+            }
+            finally
+            {
+                messageOperationsEventSource.PublishStop(outgoingContext.MessageId, isFaulted);
+            }
         }
 
         public static Task Subscribe(IBehaviorContext context, Type eventType, SubscribeOptions options)
@@ -99,7 +118,26 @@ static Task SendMessage(this IBehaviorContext context, Type messageType, object
                 outgoingContext.AddDeliveryConstraint(options.DelayedDeliveryConstraint);
             }
 
-            return outgoingContext.InvokePipeline<IOutgoingSendContext>();
+            return !MessageOperationsEventSource.Log.IsEnabled() ? outgoingContext.InvokePipeline<IOutgoingSendContext>() : InvokePipelineAndEmitEvents(outgoingContext);
+        }
+
+        static async Task InvokePipelineAndEmitEvents(IOutgoingSendContext outgoingContext)
+        {
+            var isFaulted = false;
+            var messageOperationsEventSource = MessageOperationsEventSource.Log;
+            try
+            {
+                messageOperationsEventSource.SendStart(outgoingContext.MessageId);
+                await outgoingContext.InvokePipeline().ConfigureAwait(false);
+            }
+            catch
+            {
+                isFaulted = true;
+            }
+            finally
+            {
+                messageOperationsEventSource.SendStop(outgoingContext.MessageId, isFaulted);
+            }
         }
 
         public static Task Reply(IBehaviorContext context, object message, ReplyOptions options)
@@ -132,7 +170,26 @@ static Task ReplyMessage(this IBehaviorContext context, Type messageType, object
                 options.Context,
                 context);
 
-            return outgoingContext.InvokePipeline<IOutgoingReplyContext>();
+            return !MessageOperationsEventSource.Log.IsEnabled() ? outgoingContext.InvokePipeline<IOutgoingReplyContext>() : InvokePipelineAndEmitEvents(outgoingContext);
+        }
+
+        static async Task InvokePipelineAndEmitEvents(IOutgoingReplyContext outgoingContext)
+        {
+            var isFaulted = false;
+            var messageOperationsEventSource = MessageOperationsEventSource.Log;
+            try
+            {
+                messageOperationsEventSource.ReplyStart(outgoingContext.MessageId);
+                await outgoingContext.InvokePipeline().ConfigureAwait(false);
+            }
+            catch
+            {
+                isFaulted = true;
+            }
+            finally
+            {
+                messageOperationsEventSource.ReplyStop(outgoingContext.MessageId, isFaulted);
+            }
         }
     }
 }
diff --git a/src/NServiceBus.Core/Unicast/MessageOperationsEventSource.cs b/src/NServiceBus.Core/Unicast/MessageOperationsEventSource.cs
@@ -0,0 +1,66 @@
+﻿namespace NServiceBus
+{
+    using System;
+    using System.Diagnostics.Tracing;
+
+    /// <summary>
+    /// Do not rename the methods or parameters on this class nor change the event declarations because it directly affects
+    /// the events that are generated.
+    /// </summary>
+    /// <remarks>
+    /// Operations here are used on on hot path where async state machine was optimized away. Enable check done as
+    /// part of caller
+    /// </remarks>
+    [EventSource(Name = EventSourceName)]
+    sealed class MessageOperationsEventSource : EventSource
+    {
+        MessageOperationsEventSource()
+        {
+        }
+
+        [Event(SendStartEventId, Message = "Sending message with MessageId '{0}' started.", Level = EventLevel.Informational)]
+        public void SendStart(string MessageId) => WriteEvent(SendStartEventId, MessageId);
+
+        [Event(SendStopEventId, Message = "Sending message with MessageId '{0}' stopped.", Level = EventLevel.Informational)]
+        public void SendStop(string MessageId, bool IsFaulted) => WriteEvent(SendStopEventId, MessageId, IsFaulted);
+
+        [Event(PublishStartEventId, Message = "Publishing message with MessageId '{0}' started.", Level = EventLevel.Informational)]
+        public void PublishStart(string MessageId) => WriteEvent(PublishStartEventId, MessageId);
+
+        [Event(PublishStopEventId, Message = "Publishing message with MessageId '{0}' stopped.", Level = EventLevel.Informational)]
+        public void PublishStop(string MessageId, bool IsFaulted) => WriteEvent(PublishStopEventId, MessageId, IsFaulted);
+
+        [Event(ReplyStartEventId, Message = "Replying message with MessageId '{0}' started.", Level = EventLevel.Informational)]
+        public void ReplyStart(string MessageId) => WriteEvent(ReplyStartEventId, MessageId);
+
+        [Event(ReplyStopEventId, Message = "Replying message with MessageId '{0}' stopped.", Level = EventLevel.Informational)]
+        public void ReplyStop(string MessageId, bool IsFaulted) => WriteEvent(ReplyStopEventId, MessageId, IsFaulted);
+
+        // optimized version for the common signature
+        [NonEvent]
+        unsafe void WriteEvent(int EventId, string MessageId, bool IsFaulted)
+        {
+            fixed(char* messageIdPtr = MessageId)
+            {
+                var eventPayload = stackalloc EventData[2];
+
+                eventPayload[0].Size = (MessageId.Length + 1) * 2;
+                eventPayload[0].DataPointer = (IntPtr)messageIdPtr;
+
+                eventPayload[1].Size = sizeof(bool);
+                eventPayload[1].DataPointer = (IntPtr)(&IsFaulted);
+                WriteEventCore(EventId, 2, eventPayload);
+            }
+        }
+
+        const string EventSourceName = "NServiceBus.Messages";
+        const int SendStartEventId = 1;
+        const int SendStopEventId = 2;
+        const int PublishStartEventId = 3;
+        const int PublishStopEventId = 4;
+        const int ReplyStartEventId = 5;
+        const int ReplyStopEventId = 6;
+
+        internal static readonly MessageOperationsEventSource Log = new MessageOperationsEventSource();
+    }
+}
diff --git a/src/NServiceBus.Testing.Fakes/AsyncLocal.cs b/src/NServiceBus.Testing.Fakes/AsyncLocal.cs

Original file line number	Diff line number	Diff line change
`@@ -25,8 +25,12 @@ public async Task Invoke(MessageContext messageContext)`
`25`	`25`	`var rootContext = pipelineComponent.CreateRootContext(childBuilder, messageContext.Extensions);`
`26`	`26`	`var transportReceiveContext = new TransportReceiveContext(message, messageContext.TransportTransaction, messageContext.ReceiveCancellationTokenSource, rootContext);`
`27`	`27`
	`28`	`+ var isFaulted = false;`
	`29`	`+ var pipelineEventSource = PipelineEventSource.Log;`
`28`	`30`	`try`
`29`	`31`	`{`
	`32`	`+ pipelineEventSource.MainStart(messageContext.MessageId);`
	`33`	`+`
`30`	`34`	`await transportReceiveContext.InvokePipeline<ITransportReceiveContext>().ConfigureAwait(false);`
`31`	`35`	`}`
`32`	`36`	`catch (Exception e)`
`@@ -39,6 +43,10 @@ public async Task Invoke(MessageContext messageContext)`
`39`	`43`
`40`	`44`	`throw;`
`41`	`45`	`}`
	`46`	`+ finally`
	`47`	`+ {`
	`48`	`+ pipelineEventSource.MainStop(messageContext.MessageId, isFaulted);`
	`49`	`+ }`
`42`	`50`
`43`	`51`	`await transportReceiveContext.RaiseNotification(new ReceivePipelineCompleted(message, pipelineStartedAt, DateTime.UtcNow)).ConfigureAwait(false);`
`44`	`52`	`}`