How to log Correlation Id with Serilog when using MassTransit as a Distributed log?

[berkslv]

I have a architectural question about distributed logging, and I can't see any issue or discussion around this topic, so I created this discussion.

If you have n services that implement microservice architecture and communicates with other services from HTTP calls and event queues such as RabbitMQ (with masstransit), how can you log CorrelationId in this communicational flow between services? Is there a solution for that or we should implement our custom code?

I think something like that as a solution but this is open to any kind of question. Enricher part won't work right now, I will implement ILogEventEnricher in the right way if this approach works.

In summary, when the service receives a request from HTTP, CorrelationMiddleware will set the correlationId to the Correlation class. When the service receives a request from RabbitMQ via MassTransit, CorrelationConsumeFilter will set the correlationId to the Correlation class. When you publish or send an event to the queue, CorrelationPublishFilter or CorrelationSendFilter will handle the task and set the CorrelationId field in PublishContext or SendContext for the event's consumer to retrieve the correlation id. If you make an HTTP call to any service, you have to manually set the CorrelationId to be received by the receiving application.

namespace Notification.Api.Domain.ServiceModels
{
    public class Correlation
    {
        public Guid? CorrelationId { get; set; }
    }
}

namespace Notification.Api.Middlewares
{

    public class CorrelationEnricher : ILogEventEnricher
    {
        private readonly Correlation _correlation;

        public CorrelationEnricher(Correlation correlation)
        {
            _correlation = correlation;
        }

        public void Enrich(LogEvent logEvent, ILogEventPropertyFactory pf)
        {
            if (_correlation.CorrelationId is not null)
            {
                string correlationId = _correlation.CorrelationId.Value.ToString();
                LogEventProperty property = new LogEventProperty("CorrelationId", new ScalarValue(correlationId));
                logEvent.AddOrUpdateProperty(property);
            }
        }
    }

    public class CorrelationMiddleware
    {
        private readonly RequestDelegate _next;
        public CorrelationMiddleware(RequestDelegate next)
        {
            _next = next;
        }
        public async Task InvokeAsync(HttpContext context, Correlation correlation)
        {
            var correlationId = context.Request.Headers["CorrelationId"];
            if (!string.IsNullOrWhiteSpace(correlationId))
            {
                correlation.CorrelationId = Guid.Parse(correlationId.ToString());
            }

            await _next(context);
        }
    }

    public class CorrelationConsumeFilter<T> : IFilter<ConsumeContext<T>> where T : class
    {
        private readonly Correlation _correlation;
        private readonly IHttpContextAccessor _httpContextAccessor;

        public CorrelationConsumeFilter(Correlation correlation, IHttpContextAccessor httpContextAccessor)
        {
            _correlation = correlation;
            _httpContextAccessor = httpContextAccessor;
        }

        public Task Send(ConsumeContext<T> context, IPipe<ConsumeContext<T>> next)
        {
            var correlationId = context.CorrelationId;

            if (correlationId.HasValue)
            {
                _correlation.CorrelationId = correlationId.Value;
                var httpContext = _httpContextAccessor.HttpContext;

                if (httpContext is not null)
                {
                    httpContext.Request.Headers["CorrelationId"] = correlationId.Value.ToString();
                }
            }

            return next.Send(context);
        }

        public void Probe(ProbeContext context)
        {
        }
    }

    public class CorrelationPublishFilter<T> : IFilter<PublishContext<T>> where T : class
    {
        readonly Correlation _correlation;

        public CorrelationPublishFilter(Correlation correlation)
        {
            _correlation = correlation;
        }

        public Task Send(PublishContext<T> context, IPipe<PublishContext<T>> next)
        {
            if (_correlation.CorrelationId is not null)
            {
                context.CorrelationId = _correlation.CorrelationId;
            }


            return next.Send(context);
        }

        public void Probe(ProbeContext context)
        {
        }
    }

    public class CorrelationSendFilter<T> : IFilter<SendContext<T>> where T : class
    {
        readonly Correlation _correlation;

        public CorrelationSendFilter(Correlation correlation)
        {
            _correlation = correlation;
        }

        public Task Send(SendContext<T> context, IPipe<SendContext<T>> next)
        {
            if (_correlation.CorrelationId is not null)
            {
                context.CorrelationId = _correlation.CorrelationId;
            }

            return next.Send(context);
        }

        public void Probe(ProbeContext context)
        {
        }
    }

    
}

Program.cs

var builder = WebApplication.CreateBuilder(args);

// ...

builder.Services.AddHttpContextAccessor();

builder.Services.AddScoped<Correlation>();

var messageBroker = builder.Configuration.GetSection("MessageBroker");
builder.Services.AddMassTransit(cfg =>
{
    cfg.SetKebabCaseEndpointNameFormatter();

    cfg.AddConsumer<SendSmsConsumer>();
    cfg.AddConsumer<SendMailConsumer>();

    cfg.UsingRabbitMq((context, config) =>
    {
        config.UseSendFilter(typeof(CorrelationSendFilter<>), context);
        config.UsePublishFilter(typeof(CorrelationPublishFilter<>), context);
        config.UseConsumeFilter(typeof(CorrelationConsumeFilter<>), context);


        config.UseMessageRetry(r => r.Immediate(5));

        config.Host(messageBroker["Host"], messageBroker["VirtualHost"], h =>
        {
            h.Username(messageBroker["Username"]!);
            h.Password(messageBroker["Password"]!);
        });

        config.ConfigureEndpoints(context);
    });
});


var index = $"{builder.Configuration.GetValue<string>("ProjectName")}-{DateTime.Today:yyyyMMdd}";
Log.Logger = new LoggerConfiguration()
            .Enrich.With<CorrelationEnricher>()
            .ReadFrom.Configuration(builder.Configuration)
            .CreateLogger();

builder.Host.UseSerilog();

var app = builder.Build();

// ...

app.UseMiddleware<CorrelationMiddleware>();

// ...

app.Run();