Merge pull request #31 from Particular/fix-metrics

Metrics fixes

Author: dvdstelt

src/NServiceBus.Metrics/CriticalTime/CriticalTimeMetricBuilder.cs

@@ -7,8 +7,20 @@
 
 class CriticalTimeMetricBuilder : MetricBuilder
 {
+    readonly ResetMetricTimer resetMetricTimer;
+
+    public CriticalTimeMetricBuilder(ResetMetricTimer resetMetricTimer)
+    {
+        this.resetMetricTimer = resetMetricTimer;
+    }
+
     public override void WireUp(FeatureConfigurationContext featureConfigurationContext)
     {
+        resetMetricTimer.NoMessageSentForAWhile += (sender, message) =>
+        {
+            criticalTimeTimer.Record(0, TimeUnit.Milliseconds);
+        };
+
         featureConfigurationContext.Pipeline.OnReceivePipelineCompleted(e =>
         {
             DateTime timeSent;
@@ -23,6 +35,6 @@ public override void WireUp(FeatureConfigurationContext featureConfigurationCont
         });
     }
 
-    [Timer("Critical Time", "Messages", "Age of the oldest message in the queue.")]
+    [Timer("Critical Time", "Messages", "The time it took from sending to processing the message.")]
     Timer criticalTimeTimer = default(Timer);
 }

src/NServiceBus.Metrics/MetricsFeature.cs

@@ -17,9 +17,11 @@ protected override void Setup(FeatureConfigurationContext context)
     {
         context.ThrowIfSendonly();
 
-        context.RegisterMetricBuilder(new CriticalTimeMetricBuilder());
+        var resetMetricTimer = new ResetMetricTimer(context);
+
+        context.RegisterMetricBuilder(new CriticalTimeMetricBuilder(resetMetricTimer));
         context.RegisterMetricBuilder(new PerformanceStatisticsMetricBuilder());
-        context.RegisterMetricBuilder(new ProcessingTimeMetricBuilder());
+        context.RegisterMetricBuilder(new ProcessingTimeMetricBuilder(resetMetricTimer));
         context.RegisterMetricBuilder(new QueueLengthMetricBuilder());
 
         var settings = context.Settings;

src/NServiceBus.Metrics/NServiceBus.Metrics.csproj

@@ -86,6 +86,7 @@
     <Compile Include="Reporters\TraceReport.cs" />
     <Compile Include="AssemblyInfo.cs" />
     <Compile Include="InternalsVisibleTo.cs" />
+    <Compile Include="ResetMetricTimer.cs" />
   </ItemGroup>
   <ItemGroup>
     <None Include="NServiceBus.snk" />

src/NServiceBus.Metrics/ProcessingTime/ProcessingTimeMetricBuilder.cs

@@ -6,8 +6,20 @@
 
 class ProcessingTimeMetricBuilder : MetricBuilder
 {
+    readonly ResetMetricTimer resetMetricTimer;
+
+    public ProcessingTimeMetricBuilder(ResetMetricTimer resetMetricTimer)
+    {
+        this.resetMetricTimer = resetMetricTimer;
+    }
+
     public override void WireUp(FeatureConfigurationContext featureConfigurationContext)
     {
+        resetMetricTimer.NoMessageSentForAWhile += (sender, message) =>
+        {
+            processingTimeTimer.Record(0, TimeUnit.Milliseconds);
+        };
+
         featureConfigurationContext.Pipeline.OnReceivePipelineCompleted(e =>
         {
             var processingTimeInMilliseconds = ProcessingTimeCalculator.Calculate(e.StartedAt, e.CompletedAt).TotalMilliseconds;

src/NServiceBus.Metrics/ResetMetricTimer.cs

@@ -0,0 +1,47 @@
+using System;
+using System.Threading.Tasks;
+using NServiceBus;
+using NServiceBus.Features;
+
+class ResetMetricTimer
+{
+    public event EventHandler NoMessageSentForAWhile;
+
+    public ResetMetricTimer(FeatureConfigurationContext featureConfigurationContext)
+    {
+        timer = new System.Threading.Timer(ResetCounterValueIfNoMessageHasBeenProcessedRecently, null, 0, 2000);
+
+        featureConfigurationContext.Pipeline.OnReceivePipelineCompleted(e =>
+        {
+            DateTime timeSent;
+            if (e.TryGetTimeSent(out timeSent))
+            {
+                lastMessageProcessedTime = e.CompletedAt;
+            }
+
+            return Task.FromResult(0);
+        });
+
+    }
+
+    void ResetCounterValueIfNoMessageHasBeenProcessedRecently(object state)
+    {
+        // Concurreny : This runs every 2s and might just fire before an update
+        if (NoMessageHasBeenProcessedRecently())
+        {
+            NoMessageSentForAWhile?.Invoke(this, null);
+        }
+    }
+
+    bool NoMessageHasBeenProcessedRecently()
+    {
+        var timeFromLastMessageProcessed = DateTime.UtcNow - lastMessageProcessedTime;
+        // Concurrency : This currently ignores messages being processed atm.
+        return timeFromLastMessageProcessed > estimatedMaximumProcessingDuration;
+    }
+
+    TimeSpan estimatedMaximumProcessingDuration = TimeSpan.FromSeconds(2);
+    DateTime lastMessageProcessedTime;
+    // ReSharper disable once NotAccessedField.Local
+    System.Threading.Timer timer; // We need this member or GC will dispose timer immediately.
+}