Azure ServiceBus Receive Transaction not accessible as `CurrentTransaction`

[lenaschoenburg]

APM Agent version

1.10.0

Environment

Operating system and version: Linux

.NET Framework/Core name and version: .NET 5

Application Target Framework: ASP.NET Core Runtime 5.0.6

Describe the bug

With the new Azure ServiceBus support, apm-agent correctly creates transactions for every received message.
However, the transaction does not contain any spans and creating spans manually fails because there is no CurrentTransaction.

Expected behavior

Calling Tracer.CurrentTransaction() in a message handler returns the automatically created transaction.

Actual behavior

Calling Tracer.CurrentTransaction() in a message handler returns null.

[lenaschoenburg]

Creating a transaction manually does work and spans (for example from EF Core queries) are collected.

[gregkalapos]

@russcam will know more about this, nevertheless a few ideas here:

• I think the reason for the CurrentTransaction being null is that we stop the transaction when one of the Stop events from ServiceBus is fired and this event is fired (I guess) once reading the message is done.
• If you start the transaction manually before reading from ServiceBus happens, then instead of a transaction, the agent will create a ServiceBus span on the manually created transaction. This way you'll be able to capture the ServiceBus event and the other spans as well. Not ideal, but may help.

[lenaschoenburg]

@gregkalapos Thanks for your reply!

I think the reason for the CurrentTransaction being null is that we stop the transaction when one of the Stop events from ServiceBus is fired and this event is fired (I guess) once reading the message is done

I don't think this is the cause because the automatically created transaction has a duration that correlates to the time it takes for our message handler to run (>10s). So there should be a CurrentTransaction when handling the message.

If you start the transaction manually before reading from ServiceBus happens

This won't work for my use case where we create a ServiceBusSessionProcessor that is staying open and continuously pulling new messages, calling the message handler for each.

[gregkalapos]

I don't think this is the cause because the automatically created transaction has a duration that correlates to the time it takes for our message handler to run (>10s). So there should be a CurrentTransaction when handling the message.

Great, that's a good sign, maybe we can make this work. @oleschoenburg how do you get the ITracer instance where you see null current transaction? And how do you enable the agent?

[lenaschoenburg]

I enable the agent with IApplicationBuilder.UseAllElasticApm() and try to get the current transaction via the static Tracer: var tx = Elastic.Apm.Agent.Tracer.CurrentTransaction;

[gregkalapos]

Could it be that IApplicationBuilder.UseAllElasticApm() is not executed before you call Elastic.Apm.Agent.Tracer.CurrentTransaction?

Some ideas here:

• Can you maybe check Elastic.Apm.Agent.IsConfigured before you access the static tracer?
• Also, have you tried using the IHostBuilder extension? Like here. This way you can also use DI to get the ITracer instance and that way you won't need the static API, and with that you can make sure the tracer is initialized before you access it.

[lenaschoenburg]

Could it be that IApplicationBuilder.UseAllElasticApm() is not executed before you call Elastic.Apm.Agent.Tracer.CurrentTransaction

Interesting! I don't think this is the case because this happens for every call of Elastic.Apm.Agent.Tracer.CurrentTransaction, long after IApplicationBuilder.UseAllElasticApm() was called.

I'll try using the IHostBuilder extension (though I'll have to figure out how to skip that for local test environments where we don't have access to APM).

[lenaschoenburg]

Okay, even with an ITracer from DI, CurrentTransaction returns null. It might be relevant that this is an IHostedService that is running in the background.

[gregkalapos]

Thanks @oleschoenburg for the info - I'm out of ideas for now. Maybe @russcam will know.

If we can't reproduce, then likely we'll ping you here with more questions.

[lenaschoenburg]

Here is a reproducer: https://gist.github.com/oleschoenburg/021b0e25992a4e1249204949a90bdccb

For me this throws an exception for every message on the subscription.

[russcam]

Thanks for opening @oleschoenburg.

The ServiceBus integration is achieved through a diagnostic listener. The diagnostic events for Azure.Messaging.ServiceBus are

Operation Name	Tracked API
ServiceBusSender.Send	ServiceBusSender.SendMessageAsync ServiceBusSender.SendMessagesAsync
ServiceBusSender.Schedule	ServiceBusSender.ScheduleMessageAsync ServiceBusSender.ScheduleMessagesAsync
ServiceBusSender.Cancel	ServiceBusSender.CancelScheduledMessageAsync ServiceBusSender.CancelScheduledMessagesAsync
ServiceBusReceiver.Receive	ServiceBusReceiver.ReceiveMessageAsync ServiceBusReceiver.ReceiveMessagesAsync
ServiceBusReceiver.ReceiveDeferred	ServiceBusReceiver.ReceiveDeferredMessagesAsync
ServiceBusReceiver.Peek	ServiceBusReceiver.PeekMessageAsync ServiceBusReceiver.PeekMessagesAsync
ServiceBusReceiver.Abandon	ServiceBusReceiver.AbandonMessagesAsync
ServiceBusReceiver.Complete	ServiceBusReceiver.CompleteMessagesAsync
ServiceBusReceiver.DeadLetter	ServiceBusReceiver.DeadLetterMessagesAsync
ServiceBusReceiver.Defer	ServiceBusReceiver.DeferMessagesAsync
ServiceBusReceiver.RenewMessageLock	ServiceBusReceiver.RenewMessageLockAsync
ServiceBusSessionReceiver.RenewSessionLock	ServiceBusSessionReceiver.RenewSessionLockAsync
ServiceBusSessionReceiver.GetSessionState	ServiceBusSessionReceiver.GetSessionStateAsync
ServiceBusSessionReceiver.SetSessionState	ServiceBusSessionReceiver.SetSessionStateAsync
ServiceBusProcessor.ProcessMessage	Processor callback set on ServiceBusProcessor. ProcessMessageAsync property
ServiceBusSessionProcessor.ProcessSessionMessage	Processor callback set on ServiceBusSessionProcessor. ProcessMessageAsync property

At the moment, the integration listens only for the following events, based on the messaging and azure specs

Operation Name
ServiceBusSender.Send
ServiceBusSender.Schedule
ServiceBusReceiver.Receive
ServiceBusReceiver.ReceiveDeferred

We should consider instrumenting more events, particularly ServiceBusProcessor.ProcessMessage and ServiceBusSessionProcessor.ProcessSessionMessage that are emitted by ServiceBusProcessor and ServiceBusSessionProcessor, respectively.

With ServiceBusProcessor.ProcessMessage and ServiceBusSessionProcessor.ProcessSessionMessage listened to however, it looks like there is still a challenge; there doesn't appear to be a single diagnostic event that encapsulates all of the actions that we might want to capture in a single transaction. Consider the following setup based on your gist (thank you for this!);

using System;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Azure.Messaging.ServiceBus;
using Azure.Messaging.ServiceBus.Administration;
using Microsoft.Extensions.Azure;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace ApmAgentDotnet1321 
{
    internal class MessageHandler : IHostedService 
    {
        private readonly ServiceBusProcessor _processor;

        public MessageHandler(ServiceBusClient client)
        {
            _processor = client.CreateProcessor(
                Environment.GetEnvironmentVariable("SB_QUEUE"));
            _processor.ProcessMessageAsync += HandleMessage;
            _processor.ProcessErrorAsync += HandleError;
        }

        private Task HandleError(ProcessErrorEventArgs args) {
            Console.WriteLine("HandleError. Exception: {0}, ErrorSource: {1}", args.Exception, args.ErrorSource);
            return Task.CompletedTask;
        }

        private Task HandleMessage(ProcessMessageEventArgs args) {
            Console.WriteLine("HandleMessage. {0}", args.Message);
            return Task.CompletedTask;
        }

        public Task StartAsync(CancellationToken cancellationToken) {
            return _processor.StartProcessingAsync(cancellationToken);
        }

        public Task StopAsync(CancellationToken cancellationToken) {
            return _processor.StopProcessingAsync(cancellationToken);
        }
    }
    
    internal static class Program
    {
        private static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .UseConsoleLifetime()
                .ConfigureLogging(loggerBuilder =>
                {
                    loggerBuilder.ClearProviders();
                })
                .ConfigureServices((hostBuilder, services) =>
                {
                    services.AddAzureClients(builder => {
                        builder.AddServiceBusClient(Environment.GetEnvironmentVariable("SB_CONN"));
                    });
                    
                    services.AddHostedService<MessageHandler>();
                });

        public static async Task<int> Main(string[] args)
        {
            var connectionString = Environment.GetEnvironmentVariable("SB_CONN");
            var queueName = Environment.GetEnvironmentVariable("SB_QUEUE");
            
            var adminClient = new ServiceBusAdministrationClient(connectionString);
            if (!(await adminClient.QueueExistsAsync(queueName)).Value)
                await adminClient.CreateQueueAsync(queueName);
            
            var client = new ServiceBusClient(connectionString);
            var sender = client.CreateSender(queueName);
            await sender.SendMessagesAsync(
                Enumerable.Range(1, 5)
                    .Select(i => new ServiceBusMessage($"test message {i}"))).ConfigureAwait(false);
            
            DiagnosticListener.AllListeners.Subscribe(listener =>
            {
                if (listener.Name == "Azure.Messaging.ServiceBus")
                {
                    listener.Subscribe(evnt =>
                    {
                        Console.WriteLine($"[{DateTime.Now:O}] {evnt.Key}");
                    });
                }
            });
            
            await CreateHostBuilder(args).Build().RunAsync();
            return 0;
        }
    }
}

This will create a queue, send 5 messages to it, then hook up a diagnostic listener to listen to events from Azure.Messaging.ServiceBus.

This prints the following to the console

[2021-06-03T15:51:24.1224734+10:00] ServiceBusReceiver.Receive.Start
[2021-06-03T15:51:26.0556184+10:00] ServiceBusReceiver.Receive.Stop
[2021-06-03T15:51:26.0574150+10:00] ServiceBusProcessor.ProcessMessage.Start
HandleMessage. {MessageId:01ed33212eb04de7a074521371d33009}
[2021-06-03T15:51:26.0647717+10:00] ServiceBusReceiver.Complete.Start
[2021-06-03T15:51:26.2837665+10:00] ServiceBusReceiver.Complete.Stop
[2021-06-03T15:51:26.2869303+10:00] ServiceBusProcessor.ProcessMessage.Stop
[2021-06-03T15:51:26.2871599+10:00] ServiceBusReceiver.Receive.Start
[2021-06-03T15:51:26.4456859+10:00] ServiceBusReceiver.Receive.Stop
[2021-06-03T15:51:26.4458528+10:00] ServiceBusProcessor.ProcessMessage.Start
HandleMessage. {MessageId:ba477e4414c842a1b120cb07e929efdf}
[2021-06-03T15:51:26.4462471+10:00] ServiceBusReceiver.Complete.Start
[2021-06-03T15:51:26.6989838+10:00] ServiceBusReceiver.Complete.Stop
[2021-06-03T15:51:26.6993898+10:00] ServiceBusProcessor.ProcessMessage.Stop
[2021-06-03T15:51:26.6997015+10:00] ServiceBusReceiver.Receive.Start
[2021-06-03T15:51:26.8579510+10:00] ServiceBusReceiver.Receive.Stop
[2021-06-03T15:51:26.8581149+10:00] ServiceBusProcessor.ProcessMessage.Start
HandleMessage. {MessageId:780e1351241c443eafece720b1926ec4}
[2021-06-03T15:51:26.8585168+10:00] ServiceBusReceiver.Complete.Start
[2021-06-03T15:51:27.0486612+10:00] ServiceBusReceiver.Complete.Stop
[2021-06-03T15:51:27.0490387+10:00] ServiceBusProcessor.ProcessMessage.Stop
[2021-06-03T15:51:27.0492999+10:00] ServiceBusReceiver.Receive.Start
[2021-06-03T15:51:27.2082718+10:00] ServiceBusReceiver.Receive.Stop
[2021-06-03T15:51:27.2085588+10:00] ServiceBusProcessor.ProcessMessage.Start
HandleMessage. {MessageId:eb755b7584d8465580b9dfea600b99cb}
[2021-06-03T15:51:27.2092990+10:00] ServiceBusReceiver.Complete.Start
[2021-06-03T15:51:27.4055052+10:00] ServiceBusReceiver.Complete.Stop
[2021-06-03T15:51:27.4058801+10:00] ServiceBusProcessor.ProcessMessage.Stop
[2021-06-03T15:51:27.4061177+10:00] ServiceBusReceiver.Receive.Start
[2021-06-03T15:51:27.5663738+10:00] ServiceBusReceiver.Receive.Stop
[2021-06-03T15:51:27.5665567+10:00] ServiceBusProcessor.ProcessMessage.Start
HandleMessage. {MessageId:688b7745a0ed4504ba0ec3c1b323d059}
[2021-06-03T15:51:27.5671772+10:00] ServiceBusReceiver.Complete.Start
[2021-06-03T15:51:27.9974828+10:00] ServiceBusReceiver.Complete.Stop
[2021-06-03T15:51:27.9978919+10:00] ServiceBusProcessor.ProcessMessage.Stop
[2021-06-03T15:51:27.9980985+10:00] ServiceBusReceiver.Receive.Start
[2021-06-03T15:52:28.0256283+10:00] ServiceBusReceiver.Receive.Exception
[2021-06-03T15:52:28.0259059+10:00] ServiceBusReceiver.Receive.Stop

The events for each message follow the pattern

ServiceBusReceiver.Receive.Start
ServiceBusReceiver.Receive.Stop
ServiceBusProcessor.ProcessMessage.Start

/** inside HandleMessage delegate **/

ServiceBusReceiver.Complete.Start
ServiceBusReceiver.Complete.Stop
ServiceBusProcessor.ProcessMessage.Stop

Ideally, I think we would want an event that fired .Start at the beginning and .Stop at the end of this chain of events, allowing the APM agent to start a transaction at the start and end it at the end. Right now, the agent will be starting a transaction at ServiceBusReceiver.Receive.Start and ending it at ServiceBusReceiver.Receive.Stop, which is why there is no current transaction inside the message handler. Listening for ServiceBusProcessor.ProcessMessage could allow a transaction to be created for message processing, which would capture more than it's currently doing, although the Receive would still be a separate transaction.

The ServiceBusProcessor writes other events through a ServiceBusEventSource, such as start processing, stop processing, renew message lock. An EventSourceLogForwarder is configured to write these to logs.

I think some more investigation is needed to ascertain what is exposed that could be used to fulfil our needs.

[lenaschoenburg]

Great explanation @russcam! I agree that a transaction spanning the entire chain from Receive.Start to ProcessMessage.Stop would be ideal. Just having a transaction from ProcessMessage.Start to ProcessMessage.Stop seems like a decent compromise though and would be good enough for my use case.