Apm-agent-dotnet: Missing Dependencies when Using Microsoft.Azure.ServiceBus

[gardnerr]

APM Agent version

<PackageReference Include="Azure.Storage.Blobs" Version="12.8.4" />
<PackageReference Include="Elastic.Apm" Version="1.11.0" />
<PackageReference Include="Elastic.Apm.Azure.Storage" Version="1.11.0" />

Environment

.NET 5.0

Describe the bug

I have a service that uses NServiceBus, which internally uses Microsoft.Azure.ServiceBus. I use UseAllElasticApm however the only thing I am seeing in APM is AzureServiceBus RECEIVE from xxxx in the transactions section and nothing in dependencies.

I think this may be a similar issue to https://github.com/elastic/apm-agent-dotnet/issues/1321, as I have managed to reproduce the issue where CurrentTransaction is null. Manually creating the transaction solves the issue but is not ideal.

Here is the code I used:

```csharp
using System;
using System.Threading;
using System.Threading.Tasks;
using Elastic.Apm.Api;
using Elastic.Apm.NetCoreAll;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Azure.ServiceBus.Core;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

namespace ApmServiceBusTest
{
internal class MessageHandler : IHostedService
{
private readonly IServiceProvider _serviceProvider;
private readonly string cnnString = "";
private readonly string queueName = "";

    public MessageHandler(IServiceProvider serviceProvider)
    {
        _serviceProvider = serviceProvider;
    }

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var receiver = new MessageReceiver(cnnString, queueName);
        using var scope = _serviceProvider.CreateScope();
        var tracer = _serviceProvider.GetService<ITracer>();

        while (true)
        {
            await receiver.ReceiveAsync();
            var transaction = tracer.CurrentTransaction;
        }
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }
}

internal class Startup
{
    public static void ConfigureServices(IServiceCollection services)
    {
        services.AddHostedService<MessageHandler>();
    }

    public static void Configure() 
    {

    }
}

internal static class Program
{
    private static IHostBuilder CreateHostBuilder(string[] args)
    {
        return Host.CreateDefaultBuilder(args)
            .ConfigureWebHostDefaults(webBuilder => { webBuilder.UseStartup<Startup>(); })
            .UseAllElasticApm();
    }

    public static void Main(string[] args)
    {
        CreateHostBuilder(args).Build().Run();
    }
}

}
`

[waczi]

Hi,

I'm experiencing the same issue. Transactions created automatically by elastic service bus instrumentation subscriber MicrosoftAzureServiceBusDiagnosticsSubscriber doesn't contain any child spans like e.g. HTTP calls. After removing library instrumentation and creating transactions explicitly in the service bus message handler everything seems to be working fine. Below I'm attaching screenshots that shows timelines for those two approaches.

Transaction created automatically by MicrosoftAzureServiceBusDiagnosticsSubscriber

Transaction created explicitly in the code that handles Service Bus message Agent.Tracer.CaptureTransaction(...)

Versions:

Elastic.Apm.Azure.ServiceBus 1.11.1
Microsoft.Azure.ServiceBus 5.1.3
Elastic stack 7.14.1

Regards.

[gregkalapos]

Thanks for reporting this.

I started looking at this, but I'm not sure I get the full picture here. I think https://github.com/elastic/apm-agent-dotnet/pull/1331 addressed this issue which was released in version 1.11.0.

Regarding the sample code from @gardnerr

 public async Task StartAsync(CancellationToken cancellationToken)
        {
            var receiver = new MessageReceiver(cnnString, queueName);
            using var scope = _serviceProvider.CreateScope();
            var tracer = _serviceProvider.GetService<ITracer>();

            while (true)
            {
                await receiver.ReceiveAsync();
                var transaction = tracer.CurrentTransaction;
            }
        }

The var transaction = tracer.CurrentTransaction; line after await receiver.ReceiveAsync(); will typically return null and I'm not sure if we can address that - this is expected. This is because after the await reading the message is over, so the agent will stop the transaction. What https://github.com/elastic/apm-agent-dotnet/pull/1331 addressed is that you can add a message handler and within that handler the agent will maintain the current transation. Like this:

public async Task StartAsync(CancellationToken cancellationToken)
{
    var receiver = new MessageReceiver(cnnString, queueName);
    receiver.RegisterMessageHandler((message, token) =>
    {
        var tracer = _serviceProvider.GetService<ITracer>();
                // Here the transaction is not null, code here will be part of the transaction
        var transaction = tracer.CurrentTransaction;
        return Task.CompletedTask;
    }, args =>
    {
        return Task.CompletedTask;
    });
    using var scope = _serviceProvider.CreateScope();
    var tracer = _serviceProvider.GetService<ITracer>();
    while (true)
    {
        await receiver.ReceiveAsync();
                // at this point reading is over, so this is expected to be null
        var transaction = tracer.CurrentTransaction;
    }
}

Now @waczi says

After removing library instrumentation and creating transactions explicitly in the service bus message handler everything seems to be working fine.

Could you maybe give us a small reproducer of this?

So to sum this up: from the code sample I see here, everything woks as expected and I just tested the scenario where I added a message handler and that seems to maintain the current span - we also have tests we run in our CI covering this. You can see those here and here. Is this maybe something which is not covered?

Another reproducer would be super helpful.

[gregkalapos]

Also for completeness: we got another reproducer with NServiceBus over Azure ServiceBus where the current transaction was null within an IHandleMessages implementation. I think this is from @gardnerr, but maybe I'm wrong - I'll also reach out through other channels.

I debugged through this and this is an NServiceBus related issue: the underlaying Azure ServiceBus library sends a .Stop event before the IHandleMessages implementation is called and this causes the agent to end the transaction before the IHandleMessages implementation runs. So this is kind of an unfortunate combination. We don't support NServiceBus directly, we support the underlaying Azure ServiceBus library and the way NServiceBus runs its message handlers (specifically the IHandleMessages implementations) is that those run after the reading from Azure ServiceBus is alrerady closed. I think for this to work out of the box, we'd need specific support for NServiceBus.

[waczi]

Hi @gregkalapos ,

below you can find some code, and screenshots.

• Startup.cs implementation. Note that I put there elastic configuration for both approaches just to not paste almost the same code twice.

public class Startup
{
    public Startup(IConfiguration configuration)
    {
        Configuration = configuration;
    }

    private IConfiguration Configuration { get; }

    public void ConfigureServices(IServiceCollection services)
    {
        services.AddControllers();
    }

    public void Configure(
        IApplicationBuilder app,
        IWebHostEnvironment env,
        IHostApplicationLifetime lifeTime)
    {
        RegisterApplicationStartedEvents(lifeTime);

        if (env.IsDevelopment())
        {
            app.UseDeveloperExceptionPage();
        }

        app.UseRouting();
        app.UseAuthorization();
        app.UseEndpoints(endpoints =>
        {
            endpoints.MapControllers();
        });

        //Using MicrosoftAzureServiceBusDiagnosticsSubscriber
        app.UseElasticApm(
            Configuration,
            new AspNetCoreDiagnosticSubscriber(),
            new HttpDiagnosticsSubscriber(),
            new MicrosoftAzureServiceBusDiagnosticsSubscriber());

        //Not using MicrosoftAzureServiceBusDiagnosticsSubscriber
        app.UseElasticApm(
            Configuration,
            new AspNetCoreDiagnosticSubscriber(),
            new HttpDiagnosticsSubscriber());
    }

    private static void RegisterApplicationStartedEvents(IHostApplicationLifetime lifeTime)
    {
        var subscriptionClient = new SubscriptionClient("***", "test", "test");
        var messageReceiver = new ServiceBusMessageReceiver(new HttpClient());

        lifeTime.ApplicationStarted.Register(() =>
        {
            subscriptionClient.RegisterMessageHandler(
                messageReceiver.HandleMessageAsync,
                new MessageHandlerOptions(
                    ServiceBusMessageReceiver.HandleExceptionAsync)
                {
                    AutoComplete = false,
                    MaxConcurrentCalls = 1
                });
        });
    }
}

• appsettings.json ElasticmApm section

"ElasticApm": {
  "ServerUrl": "<apm-elastic-url>",
  "ServiceName": "<service-name>",
  "ServiceVersion": "1.0.0",
  "EnableLogCorrelation": true,
  "CloudProvider": "azure",
  "FlushInterval": "0s",
  "MetricsInterval": "1s",
  "SpanFramesMinDuration": "-1ms"
}

• Service bus message receiver implementation that is basing on MicrosoftAzureServiceBusDiagnosticsSubscriber

public class ServiceBusMessageReceiver
{
    private readonly HttpClient _httpClient;

    public ServiceBusMessageReceiver(HttpClient httpClient)
    {
        _httpClient = httpClient;
    }

    public async Task HandleMessageAsync(Message message, CancellationToken token)
    {
        Console.WriteLine($"Received: {Encoding.UTF8.GetString(message.Body)}");
        await _httpClient.GetAsync("https://google.com", token);
    }

    public static Task HandleExceptionAsync(ExceptionReceivedEventArgs exception)
    {
        return Task.CompletedTask;
    }
}

• Service bus message receiver implementation that is explicitly capturing new transaction for each message that is received from the subscription.

public class ServiceBusMessageReceiver
{
    private readonly HttpClient _httpClient;

    public ServiceBusMessageReceiver(HttpClient httpClient)
    {
        _httpClient = httpClient;
    }

    public async Task HandleMessageAsync(Message message, CancellationToken token)
    {
        var transaction = Agent.Tracer.StartTransaction("RECEIVED ServiceBus Message", "service-bus");

        try
        {
            Console.WriteLine($"Received: {Encoding.UTF8.GetString(message.Body)}");
            await _httpClient.GetAsync("https://google.com", token);
        }
        finally
        {
            transaction.End();
        }
    }

    public static Task HandleExceptionAsync(ExceptionReceivedEventArgs exception)
    {
        return Task.CompletedTask;
    }
}

• Versions:
  .NET 5
  Elastic.Apm.AspNetCore - 1.11.1
  Elastic.Apm.Azure.ServiceBus - 1.11.1
  Microsoft.Azure.ServiceBus - 5.1.3

As you can see the child http span is not tracked in case of using MicrosoftAzureServiceBusDiagnosticsSubscriber instrumentation while transaction created explicitly tracks child span. Let me know if you need any additional details.

Regards.

[gregkalapos]

@russcam will know more about this issue. Nevertheless I document my findings here.

On the screenshot above I see a transaction with name AzureServiceBus RECEIVE from [queuename]. From what I understand the handler (which is registered in subscriptionClient.RegisterMessageHandler) should be part of the Microsoft.Azure.ServiceBus.Process event and we name transactions AzureServiceBus PROCESS from [queuename] when that event is raised.

So this makes me think that there should be another transaction with AzureServiceBus PROCESS from [queuename] and the outgoing HTTP call should be part of that.

This works for me:

@waczi do you have such a transaction, or you only see the one with AzureServiceBus RECEIVE from.. (the one you pasted a screenshot from) in case you use MicrosoftAzureServiceBusDiagnosticsSubscriber?

Now - one thing I managed to reproduce is that I got 2 transactions for a single ReceiveAsync() call when I read massage, one with RECEIVE and one with PROCESS - I also debugged AzureMessagingServiceBusDiagnosticListener and for the code above we got multiple events and indeed we start transactions for multiple ones without any check. I think this is not ok - to handle this, one thing we could do is that in the diagnostic listener we check for existing transactions and not start a second one if a previous Microsoft.Azure.ServiceBus.X event was already raised. Question is what transaction name should we use?

I only tested with Microsoft.Azure.ServiceBus (the old one), I don't know how Azure.Messaging.ServiceBus behaves.

[russcam]

Thanks for opening @gardnerr, and thanks for your input, @waczi. Let me see if I can clear up how the Azure Service Bus integration works.

What you're describing @gardnerr sounds like the expected behaviour and the same as https://github.com/elastic/apm-agent-dotnet/issues/1321. It'd be good to discuss whether this is serving you and other users well though. Given this has come up a couple of times, I'd like to see if we can improve things.

The Azure Service Bus integration follows the APM messaging spec. The _intent_ of message reception is to capture a transaction around the message handling flow. When using a processor or session processor like ServiceBusProcessor in Azure.Messaging.ServiceBus, or registering a message handler in Microsoft.Azure.ServiceBus, the message handling flow is captured in a transaction with a name starting with AzureServiceBus PROCESS from <queue/topic subscription>. I put together a sample app based on @waczi's example (happy to share if it helps), using Microsoft.Azure.ServiceBus 3.0.0, and see that the message handling transactions are captured, as is the span for the HTTP call that happens as part of the transaction:

Now, if we look at the messaging type transactions, we see there are two differently named transactions captured

The AzureServiceBus RECEIVE from <queue/topic subscription> transactions relate to receiving message(s) from Azure Service Bus; a transaction is started when the Receive Start activity is seen, and stopped when the Receive Stop activity is seen. As is, this is probably not very useful, since these activities are raised one after another, and from a user's code perspective, correlate with the ReceiveAsync() call, i.e.

```c#
await receiver.ReceiveAsync(); // Starts and ends a transaction

If not using a processor type or registering a message handler, the ideal message handling flow would be:

```c#
{
    await receiver.ReceiveAsync(); // Start a transaction

    // do some work that results in captured spans

    // end the transaction
}

Unfortunately, it's not possible to do this automatically with the activities raised, when receiving messages manually from Azure Service Bus. It may make sense to capture the ReceiveAsync() calls as a span if there is a current transaction, to allow the call to be traced when manually calling ReceiveAsync(). This would require a transaction to be manually started, which is not ideal, but perhaps better than the current implementation?

When using the processors or message handlers, it is possible to capture the messaging flow, since the activities raised by the libraries follow the pattern

Receive.Start                // start a transaction
Receive.Stop                 // end transaction
ProcessMessage.Start         // start a transaction

/** inside HandleMessage processor delegate **/

Complete.Start
Complete.Stop
ProcessMessage.Stop         // end transaction

_If_ we were to capture the ReceiveAsync() calls as a span however, we _might_ not be able to capture it when it's part of a processor or message handler, since we would need an active transaction.

[waczi]

@gregkalapos @russcam

I checked it again and it looks like you are right. I can see _PROCESS_ transaction which contains child spans as expected. When I was testing it previously I hade to miss it because I assumed that there is no separation into two transactions while handling service bus message. Sorry sorry for the misunderstanding.

Regards.