Tracing and logging within BackgroundService

[stevenmccormack]

Hi all,
I am using OpenTelemetry with Elasticsearch and Kibana in an API which is working fine. Some of the API calls are triggering a background task which is added to a queue. The main application adds items to the queue using a service. The Background service periodically checks for tasks and runs them. My Problem is that when the tasks are processed I can either have traces (which are not part of the main API request) or I can have logging. As soon as I do activity.SetParentId() there is no trace sent, but if I do not set it the logs are not associated to the main API request.

I created a simplified project for debugging the issue. Please find below the relevant parts:

Program.cs:

// Try to load user secrets
var openTelemetrySettings = builder.Configuration.GetSection("OpenTelemetrySettings").Get<OpenTelemetrySettings>();

// Setup OpenTelemetry tracing
builder.Services.AddOpenTelemetryTracing(tracerProviderBuilder => {
    tracerProviderBuilder
        .AddConsoleExporter(o => { o.Targets = ConsoleExporterOutputTargets.Console; })
        .SetResourceBuilder(Config.GetOpentelemetryResource())
        .AddSource(Config.ServiceName)
        .AddHttpClientInstrumentation(o => { o.RecordException = true; })
        .AddAspNetCoreInstrumentation(o => { o.RecordException = true; })
        .AddBackgroundWorker();

    // if user secrets have been loaded add OLTP exporter
    if (openTelemetrySettings != null && !string.IsNullOrWhiteSpace(openTelemetrySettings.EndpointUrl) && !string.IsNullOrWhiteSpace(openTelemetrySettings.AuthHeader)) {
        tracerProviderBuilder.AddOtlpExporter(options => {
            options.ExportProcessorType = ExportProcessorType.Batch;
            options.Protocol = OtlpExportProtocol.Grpc;
            options.Endpoint = new Uri(openTelemetrySettings.EndpointUrl);
            options.Headers = openTelemetrySettings.AuthHeader;
        });
    }
});

// Setup OpenTelemetry logging
builder.Logging.AddFilter(null, LogLevel.Trace);
builder.Logging.AddFilter<OpenTelemetryLoggerProvider>((category, level) => {
    // Only send log messages for application or background worker namespaces to OpenTelemetry
    if (!string.IsNullOrWhiteSpace(ns) && category.ToLower().StartsWith(ns.ToLower())) return true;
    if (category.ToLower().StartsWith("backgroundworker")) return true;
    return false;
});
builder.Logging.AddOpenTelemetry(logBuilder => {
    logBuilder.IncludeFormattedMessage = true;
    logBuilder.IncludeScopes = true;
    logBuilder.ParseStateValues = true;
    logBuilder.SetResourceBuilder(Config.GetOpentelemetryResource());
    logBuilder.AddConsoleExporter(o => o.Targets = ConsoleExporterOutputTargets.Console);
    // if user secrets have been loaded add OLTP exporter
    if (openTelemetrySettings != null && !string.IsNullOrWhiteSpace(openTelemetrySettings.EndpointUrl) && !string.IsNullOrWhiteSpace(openTelemetrySettings.AuthHeader)) {
        logBuilder.AddOtlpExporter(options => {
            options.ExportProcessorType = ExportProcessorType.Batch;
            options.Protocol = OtlpExportProtocol.Grpc;
            options.Endpoint = new Uri(openTelemetrySettings.EndpointUrl);
            options.Headers = openTelemetrySettings.AuthHeader;
        });
    }
});

Config.cs

public static class Config {
    public  const string ServiceName ="OpenTelemetryTest";

    public static TracerProviderBuilder AddBackgroundWorker(this TracerProviderBuilder builder, Action<BackgroundWorkerTracingOptions>? options = null) => builder.AddSource("BackgroundWorker");

    public static ResourceBuilder GetOpentelemetryResource() {
        var e = Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT");
        var resourceBuilder = ResourceBuilder
            .CreateDefault()
            .AddService(serviceName: ServiceName, serviceVersion: typeof(Program).Assembly.GetName().Version?.ToString(), serviceInstanceId: Environment.MachineName)
            .AddAttributes(new Dictionary<string, object>() {
                {"deployment.environment", e??"Development"},
                {"server_name", Environment.MachineName}
            })
            .AddTelemetrySdk();
        return resourceBuilder;
    }
}

Controller action:

public IActionResult Index() {
    _workerService.AddTask(new SampleTask {
        Message = "Test",
        TraceId = Activity.Current?.TraceId ?? default(ActivityTraceId),
        SpanId = Activity.Current?.SpanId ?? default(ActivitySpanId)
    });
    return Ok();
}

BackgroundWorkerService.cs

/// <summary>
/// Service for communication between API and worker
/// </summary>
public class BackgroundWorkerService {
    private readonly ILogger<BackgroundWorkerService> _logger;
    private readonly ConcurrentQueue<SampleTask> _tasks = new();

    public BackgroundWorkerService(ILogger<BackgroundWorkerService> logger) {
        _logger = logger;
    }

    public bool IsEmpty => _tasks.IsEmpty;

    public void AddTask(SampleTask task) {
        _logger.LogDebug("{Ts}|Added new Task with message|{Message}", DateTime.Now.ToString("yyyy-MM-dd hh:mm:ss"), task);
        _tasks.Enqueue(task);
    }

    public IEnumerable<SampleTask> GetTasks() {
        var tasks = new List<SampleTask>();
        var i = 0;
        while (i < 5 && !_tasks.IsEmpty) {
            ++i;
            SampleTask? result;
            if (_tasks.TryDequeue(out result)) tasks.Add(result);
        }

        return tasks;
    }
}

BackgroundWorker.cs

public class BackgroundWorker : BackgroundService {
    private readonly ILogger<BackgroundWorker> _logger;
    private readonly BackgroundWorkerService _backgroundWorkerService;
    private readonly ActivitySource Source;


    public BackgroundWorker(ILogger<BackgroundWorker> logger, BackgroundWorkerService backgroundWorkerService) {
        var assembly = typeof(BackgroundWorker).Assembly;
        var version  = assembly.GetCustomAttribute<AssemblyFileVersionAttribute>()?.Version ?? "0.0.0";
        _logger                  = logger;
        _backgroundWorkerService = backgroundWorkerService;
        Source                   = new ActivitySource("BackgroundWorker", version);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
        stoppingToken.Register(Shutdown);
        while (!stoppingToken.IsCancellationRequested) {
            while (!_backgroundWorkerService.IsEmpty) {
                foreach (var task in _backgroundWorkerService.GetTasks()) {
                    await ProcessTaskV1(task); // Not setting parent
                    await ProcessTaskV2(task); // Setting parent
                    await ProcessTaskV3(task); // Setting link
                    await Task.Delay(50, stoppingToken);
                }
            }

            await Task.Delay(2500, stoppingToken);
        }
    }

    /// <summary>
    /// Version 1 is not setting ParentId.
    /// It does generate a trace and logs but they are not shown in Kibana under the API request
    /// </summary>
    /// <param name="task"></param>
    private async Task ProcessTaskV1(SampleTask task) {
        var activity = Source.StartActivity(task.Message, ActivityKind.Internal);
        if (activity != null) {
            activity.DisplayName = "Executing simple task V1";
            activity.SetTag("simpletask.message", task.Message);
            //if (task.TraceId != default && task.SpanId != default) activity.SetParentId(task.TraceId, task.SpanId);
        }

        _logger.LogInformation("{Ts}|V1|Started task execution|{Message}", DateTime.Now.ToString("yyyy-MM-dd hh:mm:ss"), task.Message);
        await Task.Delay(500);
        _logger.LogInformation("{Ts}|V1|Finished task execution|{Message}", DateTime.Now.ToString("yyyy-MM-dd hh:mm:ss"), task.Message);
        activity?.SetTag("otel.status_code", "OK");
        activity?.Dispose();
    }

    /// <summary>
    /// Version 2 is setting ParentId.
    /// It does not generate a trace but logs are shown in Kibana under the API request
    /// </summary>
    /// <param name="task"></param>
    private async Task ProcessTaskV2(SampleTask task) {
        var activity = Source.StartActivity(task.Message, ActivityKind.Internal);
        if (activity != null) {
            activity.DisplayName = "Executing simple task V2";
            activity.SetTag("simpletask.message", task.Message);
            if (task.TraceId != default && task.SpanId != default) activity.SetParentId(task.TraceId, task.SpanId);
        }

        _logger.LogInformation("{Ts}|V2|Started task execution|{Message}", DateTime.Now.ToString("yyyy-MM-dd hh:mm:ss"), task.Message);
        await Task.Delay(500);
        _logger.LogInformation("{Ts}|V2|Finished task execution|{Message}", DateTime.Now.ToString("yyyy-MM-dd hh:mm:ss"), task.Message);
        activity?.SetTag("otel.status_code", "OK");
        activity?.Dispose();
    }

    /// <summary>
    /// Version 3 is not setting ParentId but creating a link
    /// It generates a trace and logs but they are not shown in Kibana under the API request. The link is shown and contains the trace and logs
    /// </summary>
    /// <param name="task"></param>
    private async Task ProcessTaskV3(SampleTask task) {
        var links    = new List<ActivityLink> {new(new ActivityContext(task.TraceId, task.SpanId, ActivityTraceFlags.None))};
        var activity = (task.TraceId != default && task.SpanId != default) ? Source.StartActivity("SimpleTask", ActivityKind.Internal, default(ActivityContext), links: links) : Source.StartActivity("SimpleTask", ActivityKind.Internal);
        if (activity != null) {
            activity.DisplayName = "Executing simple task V3";
            activity.SetTag("simpletask.message", task.Message);
        }

        _logger.LogInformation("{Ts}|V3|Started task execution|{Message}", DateTime.Now.ToString("yyyy-MM-dd hh:mm:ss"), task.Message);
        await Task.Delay(500);
        _logger.LogInformation("{Ts}|V3|Finished task execution|{Message}", DateTime.Now.ToString("yyyy-MM-dd hh:mm:ss"), task.Message);
        activity?.SetTag("otel.status_code", "OK");
        activity?.Dispose();
    }

    private void Shutdown() {
        if (_backgroundWorkerService.IsEmpty) {
            _logger.LogInformation($"App shutting down");
        } else {
            _logger.LogWarning($"App shutting down remaining tasks will be ignored");
        }
    }

Can anybody can tell me hot to make version two also generate a trace?

[Bjego]

Hi, I had the same issue as I ran a background job in Aspire.
You have to:

• register the source in the with tacing section when you add Opentelemetry:

builder.Services.AddOpenTelemetry()
    .WithMetrics(metrics =>
    {
        ...
    })
    .WithTracing(tracing =>
    {
        tracing.AddAspNetCoreInstrumentation()
            .AddHttpClientInstrumentation()
            .AddSource("ProducerService");
        }
    });

• in your Background job create the activity source and start the activity

public class ProducerService(...): BackgroundService
{
    public static string ActivitySourceName = "ProducerService";
    private static readonly ActivitySource activitySource = new(ActivitySourceName);

    
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
            using (var activity = activitySource.StartActivity("Producer: ProduceMessage"))
            {
            ....

Thats all