diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_handlers_are_invoked.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_handlers_are_invoked.cs new file mode 100644 index 00000000000..7cf4ce7f48f --- /dev/null +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_handlers_are_invoked.cs @@ -0,0 +1,73 @@ +namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry.Metrics; + +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using EndpointTemplates; +using NServiceBus; +using AcceptanceTesting; +using NUnit.Framework; +using Conventions = AcceptanceTesting.Customization.Conventions; + +public class When_handlers_are_invoked : OpenTelemetryAcceptanceTest +{ + const string ActiveHandlersMetric = "nservicebus.messaging.active_handlers"; + const int numberOfMessages = 5; + + [Test] + public async Task Should_report_active_handlers_gauge_that_balances_once_idle() + { + using var metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener(); + + _ = await Scenario.Define() + .WithEndpoint(b => b.CustomConfig(c => + { + c.MakeInstanceUniquelyAddressable("instanceId"); + c.LimitMessageProcessingConcurrencyTo(10); + }).When(async (session, ctx) => + { + for (var x = 0; x < numberOfMessages; x++) + { + await session.SendLocal(new OutgoingMessage()); + } + })) + .Run(); + + Assert.That(metricsListener.ReportedMeters.TryGetValue(ActiveHandlersMetric, out var net), Is.True, + $"'{ActiveHandlersMetric}' gauge should be reported"); + Assert.That(net, Is.EqualTo(0), + "increments and decrements should balance once all handlers have completed"); + + metricsListener.AssertTags(ActiveHandlersMetric, + new Dictionary + { + ["nservicebus.queue"] = Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)), + ["nservicebus.discriminator"] = "instanceId", + ["nservicebus.message_type"] = typeof(OutgoingMessage).FullName, + ["nservicebus.message_handler_type"] = typeof(EndpointWithMetrics.MessageHandler).FullName + }); + } + + public class Context : ScenarioContext + { + public int OutgoingMessagesReceived; + } + + public class EndpointWithMetrics : EndpointConfigurationBuilder + { + public EndpointWithMetrics() => EndpointSetup(); + + [Handler] + public class MessageHandler(Context testContext) : IHandleMessages + { + public Task Handle(OutgoingMessage message, IMessageHandlerContext context) + { + var messagesHandled = Interlocked.Increment(ref testContext.OutgoingMessagesReceived); + testContext.MarkAsCompleted(messagesHandled == numberOfMessages); + return Task.CompletedTask; + } + } + } + + public class OutgoingMessage : IMessage; +} diff --git a/src/NServiceBus.Core.Tests/ApprovalFiles/MeterTests.Verify_MeterAPI.approved.txt b/src/NServiceBus.Core.Tests/ApprovalFiles/MeterTests.Verify_MeterAPI.approved.txt index a572d43098f..26c3d50099f 100644 --- a/src/NServiceBus.Core.Tests/ApprovalFiles/MeterTests.Verify_MeterAPI.approved.txt +++ b/src/NServiceBus.Core.Tests/ApprovalFiles/MeterTests.Verify_MeterAPI.approved.txt @@ -1,7 +1,7 @@ { "Note": "Changes to metrics API should result in an update to NServiceBusMeter version.", "MetricsSourceName": "NServiceBus.Core.Pipeline.Incoming", - "MetricsSourceVersion": "0.2.0", + "MetricsSourceVersion": "0.3.0", "Tags": [ "error.type", "execution.result", @@ -14,6 +14,7 @@ ], "Metrics": [ "nservicebus.envelope.unwrapped => Counter", + "nservicebus.messaging.active_handlers => UpDownCounter", "nservicebus.messaging.critical_time => Histogram, Unit: s", "nservicebus.messaging.failures => Counter", "nservicebus.messaging.fetches => Counter", diff --git a/src/NServiceBus.Core/Pipeline/Incoming/IncomingPipelineMetrics.cs b/src/NServiceBus.Core/Pipeline/Incoming/IncomingPipelineMetrics.cs index b842e912a3c..61f36c73c4b 100644 --- a/src/NServiceBus.Core/Pipeline/Incoming/IncomingPipelineMetrics.cs +++ b/src/NServiceBus.Core/Pipeline/Incoming/IncomingPipelineMetrics.cs @@ -21,10 +21,11 @@ class IncomingPipelineMetrics const string RecoverabilityDelayed = "nservicebus.recoverability.delayed"; const string RecoverabilityError = "nservicebus.recoverability.error"; const string EnvelopeUnwrapping = "nservicebus.envelope.unwrapped"; + const string ActiveMessageHandlers = "nservicebus.messaging.active_handlers"; public IncomingPipelineMetrics(IMeterFactory meterFactory, string queueName, string discriminator) { - var meter = meterFactory.Create("NServiceBus.Core.Pipeline.Incoming", "0.2.0"); + var meter = meterFactory.Create("NServiceBus.Core.Pipeline.Incoming", "0.3.0"); totalProcessedSuccessfully = meter.CreateCounter(TotalProcessedSuccessfully, description: "Total number of messages processed successfully by the endpoint."); totalFetched = meter.CreateCounter(TotalFetched, @@ -45,6 +46,8 @@ public IncomingPipelineMetrics(IMeterFactory meterFactory, string queueName, str description: "Total number of messages sent to the error queue."); totalEnvelopeUnwrapping = meter.CreateCounter(EnvelopeUnwrapping, description: "Total number of unwrapping attempts by the endpoint."); + activeMessageHandlers = meter.CreateUpDownCounter(ActiveMessageHandlers, + description: "Number of message handlers currently executing."); queueNameBase = queueName; endpointDiscriminator = discriminator; @@ -243,6 +246,39 @@ public void RecordSendToErrorQueue(IRecoverabilityContext recoverabilityContext) totalSentToErrorQueue.Add(1, meterTags); } + public ActiveHandlerScope TrackHandlerInvocation(IInvokeHandlerContext context) + { + if (!activeMessageHandlers.Enabled) + { + return default; + } + + var tags = BuildActiveHandlerTags(context); + activeMessageHandlers.Add(1, tags); + return new ActiveHandlerScope(activeMessageHandlers, tags); + } + + public readonly struct ActiveHandlerScope(UpDownCounter? counter, TagList tags) : IDisposable + { + public void Dispose() => counter?.Add(-1, tags); + } + + // The increment and the matching decrement must carry identical tags so they apply to the same + // time series; building them in one place keeps the two Record methods in sync. + // The TagList is a stack struct with 8 pre-allocated slots, so adding 4 tags does not + // require a heap allocation. + static TagList BuildActiveHandlerTags(IInvokeHandlerContext context) + { + var incomingPipelineMetricTags = context.Extensions.Get(); + TagList tags; + incomingPipelineMetricTags.ApplyTags(ref tags, [ + MeterTags.QueueName, + MeterTags.EndpointDiscriminator]); + tags.Add(new KeyValuePair(MeterTags.MessageType, context.MessageMetadata.MessageType.FullName)); + tags.Add(new KeyValuePair(MeterTags.MessageHandlerType, context.MessageHandler.HandlerType.FullName)); + return tags; + } + public void EnvelopeUnwrappingSucceeded(MessageContext messageContext, IEnvelopeHandler type) => RecordEnvelopeUnwrapping(messageContext, type, true, null); public void EnvelopeUnwrappingFailed(MessageContext messageContext, IEnvelopeHandler type, Exception? exception) => RecordEnvelopeUnwrapping(messageContext, type, false, exception); void RecordEnvelopeUnwrapping(MessageContext messageContext, IEnvelopeHandler type, bool succeeded, Exception? exception) @@ -276,6 +312,7 @@ void RecordEnvelopeUnwrapping(MessageContext messageContext, IEnvelopeHandler ty readonly Counter totalDelayedRetries; readonly Counter totalSentToErrorQueue; readonly Counter totalEnvelopeUnwrapping; + readonly UpDownCounter activeMessageHandlers; string queueNameBase; string endpointDiscriminator; } \ No newline at end of file diff --git a/src/NServiceBus.Core/Pipeline/Incoming/InvokeHandlerTerminator.cs b/src/NServiceBus.Core/Pipeline/Incoming/InvokeHandlerTerminator.cs index e177dc8404d..98c22570f31 100644 --- a/src/NServiceBus.Core/Pipeline/Incoming/InvokeHandlerTerminator.cs +++ b/src/NServiceBus.Core/Pipeline/Incoming/InvokeHandlerTerminator.cs @@ -26,6 +26,7 @@ protected override async Task Terminate(IInvokeHandlerContext context) // Might as well abort before invoking the handler if we're shutting down context.CancellationToken.ThrowIfCancellationRequested(); var startTime = DateTimeOffset.UtcNow; + using var handlerScope = messagingMetricsMeters.TrackHandlerInvocation(context); try { await messageHandler