Mesajlaşmada Observability: OpenTelemetry Trace Context, Queue Depth ve Consumer Lag

Sürüm notu: Bu yazıdaki kodlar .NET 10, OpenTelemetry .NET 1.20.1, RabbitMQ 4.4, RabbitMQ.Client 7.2.2, Confluent.Kafka 2.16 (Kafka 4.2) ile derlenip denenmiştir.

Önce Tahmin Et

Bir sipariş şu yolu izliyor: POST /orders → RabbitMQ → Billing (faturayı keser) → Kafka'daki bookstore.invoices topic'i → Notification (e-posta gönderir). Servislere OpenTelemetry ekliyorsun ama Confluent.Kafka için kararlı bir instrumentation paketi yok, RabbitMQ.Client 7 için ayrı bir paket de eklemiyorsun. Tek bir sipariş için kaç trace oluşur: bir mi, iki mi, üç mü? Notification'ın log satırında, API'ye (Application Programming Interface) gelen HTTP (Hypertext Transfer Protocol) isteğinin trace id'si görünür mü?

Mesajlaşmanın en zor yanı, bir isteğin artık tek bir call stack'te değil, zamana ve servislere yayılmış olması. Messaging ve Event-Driven Architecture serisinin son yazısı bu bedelden bahsediyordu; bu yazıda onu kodla ödüyoruz: W3C (World Wide Web Consortium) traceparent'ını mesaj header'ında taşımak, producer ve consumer span'leri, log scope'unda correlation id, işlem süresi metric'i, queue depth ve consumer lag. Collector, Grafana gibi altyapı ilerideki Observability serisinin konusu; burada uygulama tarafındayız.

OpenTelemetry Kurulumu

Tek bir demo uygulaması üç rolü de oynuyor: Order API, Billing worker'ı ve Notification worker'ı. Kurulum:

builder.Services.AddOpenTelemetry()
    .ConfigureResource(r => r.AddService("bookstore-lab12"))
    .WithTracing(tracing =>
    {
        tracing
            .AddAspNetCoreInstrumentation()
            .AddSource("RabbitMQ.Client.*")
            .AddSource(MessagingTelemetry.SourceName)
            .AddProcessor(new SpanLineProcessor());
        if (builder.Configuration["OTEL_EXPORTER_OTLP_ENDPOINT"] is not null)
            tracing.AddOtlpExporter();
    })
    .WithMetrics(metrics =>
    {
        metrics.AddMeter(MessagingTelemetry.MeterName);
        if (builder.Configuration["OTEL_EXPORTER_OTLP_ENDPOINT"] is not null)
            metrics.AddOtlpExporter();
    });

Üretimde span'ler OTLP (OpenTelemetry Protocol) exporter ile bir Collector'a gider. Demoda Collector yok; onun yerine biten her span'i tek satır yazan küçük bir processor var. Çıktıyı okunur tutmak için trace ve span id'lerini kısaltıyor:

public sealed class SpanLineProcessor : BaseProcessor<Activity>
{
    public override void OnEnd(Activity a)
    {
        string parent = a.ParentSpanId == default ? "-     " : a.ParentSpanId.ToString()[..6];
        Console.WriteLine($"span trace={a.TraceId.ToString()[..8]} id={a.SpanId.ToString()[..6]} parent={parent} " +
                          $"{a.Kind,-8} {a.DisplayName}");
    }
}

RabbitMQ.Client 7: Hazır Instrumentation

RabbitMQ.Client 7, .NET'in ActivitySource API'sini doğrudan kullanıyor: RabbitMQ.Client.Publisher ve RabbitMQ.Client.Subscriber adlı iki kaynak. AddSource("RabbitMQ.Client.*") bunları dinlemeye yetiyor; ek paket yok. Publish eden kodda trace'e dair tek satır yok:

public sealed class OrderPublisher(IConnection connection) : IAsyncDisposable
{
    private readonly SemaphoreSlim _gate = new(1, 1);
    private IChannel? _channel;

    public async Task PublishAsync(OrderPlaced order, CancellationToken ct = default)
    {
        var props = new BasicProperties
        {
            MessageId = Guid.NewGuid().ToString(),
            CorrelationId = order.OrderId.ToString(),
            Type = "bookstore.order-placed.v1",
            ContentType = "application/json",
            Persistent = true
        };
        byte[] body = JsonSerializer.SerializeToUtf8Bytes(order, JsonSerializerOptions.Web);
        // IChannel thread-safe değil: tek channel'ı semaphore ile koru
        await _gate.WaitAsync(ct);
        try
        {
            _channel ??= await connection.CreateChannelAsync(new CreateChannelOptions(
                publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true), ct);
            // traceparent'ı RabbitMQ.Client 7 kendisi header'a yazar
            await _channel.BasicPublishAsync("bookstore.ordering", "order.placed", mandatory: true, props, body, ct);
        }
        finally
        {
            _gate.Release();
        }
    }

    public async ValueTask DisposeAsync()
    {
        if (_channel is not null)
            await _channel.DisposeAsync();
    }
}

Bir dinleyici varsa client, publish için bir Producer span'i açar, W3C traceparent'ını mesaj header'ına yazar ve consumer tarafında, mesaj geldiğinde bu header'dan devam eden bir Consumer span'i başlatır. Span'lere messaging.system, messaging.destination.name, messaging.rabbitmq.destination.routing_key, messaging.message.id, messaging.message.conversation_id (bizim CorrelationId), messaging.message.body.size gibi OpenTelemetry messaging semantic convention attribute'ları eklenir. Önemli bir ayrıntı: dinleyici yoksa ne span açılır ne de header yazılır. Bir önceki yazıdaki outbox relay'i OpenTelemetry'siz bir süreçte çalıştığı için yayınladığı mesajlar bu demoda parent'sız, yepyeni trace'ler olarak göründü.

Kafka'da Trace Context'i Elle Taşımak

Confluent.Kafka'nın kendisi ActivitySource yayınlamıyor; OpenTelemetry contrib'deki Kafka instrumentation paketi ise yazının yazıldığı gün hâlâ alpha (0.5.0-alpha.1). Kararlı yol, propagator'ı elle çağırmak:

public static class KafkaTracing
{
    private static readonly TextMapPropagator Propagator = Propagators.DefaultTextMapPropagator;

    public static async Task ProduceAsync<T>(IProducer<string, byte[]> producer, string topic, string key, T message)
    {
        // Confluent.Kafka'nın kararlı bir instrumentation'ı yok: producer span'ini kendimiz açıyoruz
        using Activity? activity = MessagingTelemetry.Source.StartActivity($"send {topic}", ActivityKind.Producer);
        activity?.SetTag("messaging.system", "kafka");
        activity?.SetTag("messaging.operation.type", "send");
        activity?.SetTag("messaging.destination.name", topic);

        var headers = new Confluent.Kafka.Headers();
        Propagator.Inject(new PropagationContext(activity?.Context ?? default, Baggage.Current), headers,
            static (h, name, value) => h.Add(name, Encoding.UTF8.GetBytes(value)));
        await producer.ProduceAsync(topic, new Message<string, byte[]>
        {
            Key = key,
            Value = JsonSerializer.SerializeToUtf8Bytes(message, JsonSerializerOptions.Web),
            Headers = headers
        });
    }

    public static Activity? StartProcess(ConsumeResult<string, byte[]> result)
    {
        PropagationContext parent = Propagator.Extract(default, result.Message.Headers,
            static (h, name) => h.TryGetLastBytes(name, out byte[] value) ? [Encoding.UTF8.GetString(value)] : []);
        Baggage.Current = parent.Baggage;
        Activity? activity = MessagingTelemetry.Source.StartActivity($"process {result.Topic}", ActivityKind.Consumer,
            parent.ActivityContext);
        activity?.SetTag("messaging.system", "kafka");
        activity?.SetTag("messaging.operation.type", "process");
        activity?.SetTag("messaging.destination.name", result.Topic);
        activity?.SetTag("messaging.kafka.offset", result.Offset.Value);
        return activity;
    }
}

Propagators.DefaultTextMapPropagator, OpenTelemetry SDK'sı (Software Development Kit) kurulduğunda W3C trace context ve baggage'ı birlikte taşıyan composite propagator'dur. Inject, traceparent (ve varsa tracestate, baggage) header'larını Kafka header'ına UTF-8 (Unicode Transformation Format) olarak yazar; Extract tersini yapar. Notification worker'ı her mesaj için using Activity? activity = KafkaTracing.StartProcess(result); ile span'i açıyor. Kafka'nın senkron Consume döngüsünde Activity.Current span'in ömrü boyunca bu thread'de geçerli kalır, yani log satırları da otomatik olarak doğru trace id'yi alır.

Tek Trace'te Sipariş Yolculuğu

Tek bir POST /orders isteğinin çıktısı (API yanıtı ve span satırları, bittikleri sırayla):

{"orderId":"4a9febc7-b2b1-46cd-8e03-9e87a6e97f13","traceId":"3738626c672a77c93609c10d195ae5be"}
span trace=3738626c id=946803 parent=4f2c3e Producer publish order.placed
span trace=3738626c id=4f2c3e parent=-      Server   POST /orders
span trace=3738626c id=62b35a parent=1429e7 Producer send bookstore.invoices
span trace=3738626c id=1429e7 parent=946803 Consumer deliver order.placed
span trace=3738626c id=5eac68 parent=62b35a Consumer process bookstore.invoices

Baştaki sorunun cevabı: tek trace. Beş span'in hepsi 3738626c trace'inde ve parent zinciri kesintisiz: HTTP isteği (4f2c3e) → RabbitMQ publish (946803) → Billing'de deliver (1429e7) → Kafka'ya send (62b35a) → Notification'da process (5eac68). Span'lerin bitiş sırası ile başlangıç sırası farklı: publish, HTTP yanıtından önce biter; Billing'in deliver span'i, içindeki Kafka send'i bitirmeden kapanmaz. Bir trace görüntüleyicide bu, şelale şeklinde tek bir zaman çizgisi olarak görünür ve "sipariş verildikten kaç milisaniye sonra e-posta gitti" sorusunun cevabını verir.

Consumer span'inin parent'ı producer span'i olarak seçilmiş; RabbitMQ.Client 7'nin varsayılan davranışı bu. Semantic convention'lar, özellikle batch tüketimde, consumer span'ini producer'a parent yerine "link" ile bağlamayı da tanımlar. Tek mesaj işleyen consumer'larda parent ilişkisi trace'i okumayı kolaylaştırır; binlerce mesajı tek batch'te işleyen bir consumer'da her mesaj için ayrı trace'e bağlı link'ler gerekir.

Correlation Id'yi Log Scope'una Koymak

Trace id log'lara kendiliğinden geçiyor: .NET logging, Activity.Current varken her log satırına SpanId, TraceId ve ParentId içeren bir scope ekliyor. Mesaja özgü kimlikleri biz ekliyoruz:

consumer.ReceivedAsync += async (_, ea) =>
{
    long started = Stopwatch.GetTimestamp();
    string outcome = "success";
    // TraceId/SpanId'yi logger zaten ekler; mesaja özgü kimlikleri biz ekliyoruz
    using IDisposable? scope = logger.BeginScope(new Dictionary<string, object?>
    {
        ["MessageId"] = ea.BasicProperties.MessageId,
        ["CorrelationId"] = ea.BasicProperties.CorrelationId
    });
    try
    {
        OrderPlaced order = JsonSerializer.Deserialize<OrderPlaced>(ea.Body.Span, JsonSerializerOptions.Web)!;
        var invoice = new InvoiceCreated(Guid.NewGuid(), order.OrderId, order.Total);
        logger.LogInformation("Invoice {InvoiceId} created", invoice.InvoiceId);
        await KafkaTracing.ProduceAsync(producer, "bookstore.invoices", order.OrderId.ToString(), invoice);
        await channel.BasicAckAsync(ea.DeliveryTag, false);
    }
    catch (Exception ex)
    {
        outcome = "failure";
        telemetry.Failures.Add(1, new KeyValuePair<string, object?>("messaging.destination.name", "lab12.billing"));
        logger.LogError(ex, "Billing failed");
        await channel.BasicRejectAsync(ea.DeliveryTag, requeue: false);
    }
    finally
    {
        telemetry.ProcessDuration.Record(Stopwatch.GetElapsedTime(started).TotalSeconds,
            new KeyValuePair<string, object?>("messaging.system", "rabbitmq"),
            new KeyValuePair<string, object?>("messaging.destination.name", "lab12.billing"),
            new KeyValuePair<string, object?>("outcome", outcome));
    }
};

JSON (JavaScript Object Notation) console formatter'ı IncludeScopes = true ile açık. Billing'in yazdığı log satırı:

{"EventId": 0, "LogLevel": "Information", "Category": "Bookstore.Messaging.BillingWorker", "Message": "Invoice 980f6fe7-853c-40d4-af8f-dd5b1cf6f71a created", "State": {"InvoiceId": "980f6fe7-853c-40d4-af8f-dd5b1cf6f71a", "{OriginalFormat}": "Invoice {InvoiceId} created"}, "Scopes": [{"Message": "SpanId:1429e7a7d4daeda3, TraceId:3738626c672a77c93609c10d195ae5be, ParentId:94680345bbaaa52b", "SpanId": "1429e7a7d4daeda3", "TraceId": "3738626c672a77c93609c10d195ae5be", "ParentId": "94680345bbaaa52b"}, {"Message": "System.Collections.Generic.Dictionary`2[System.String,System.Object]", "MessageId": "59816837-63b6-48e0-918b-8259b1d5b2e0", "CorrelationId": "4a9febc7-b2b1-46cd-8e03-9e87a6e97f13"}]}

Satırda API'nin trace id'si (3738626c...), Billing'in deliver span'i (1429e7...), mesajın id'si ve sipariş id'si birlikte duruyor. Log arama aracında CorrelationId ile sorgu, bir siparişle ilgili bütün servislerdeki satırları getirir; trace id ile sorgu ise tek bir isteğin zincirini. İlk denememde scope'a bir de TraceId eklemiştim; çıktıda iki kez göründü, çünkü logger onu zaten ekliyor. Structured logging'in ayrıntıları ilerideki .NET'te Logging serisinde.

Metric'ler: İşlem Süresi ve Hatalar

Trace tek bir mesajın hikâyesini anlatır; sistemin genel sağlığını metric'ler söyler. IMeterFactory ile bir Meter ve iki instrument:

public sealed class MessagingTelemetry
{
    public const string SourceName = "Bookstore.Messaging";
    public const string MeterName = "Bookstore.Messaging";
    public static readonly ActivitySource Source = new(SourceName);

    public Histogram<double> ProcessDuration { get; }
    public Counter<long> Failures { get; }

    public MessagingTelemetry(IMeterFactory meterFactory)
    {
        Meter meter = meterFactory.Create(MeterName);
        // OpenTelemetry messaging semantic convention'daki metric adı ve birimi
        ProcessDuration = meter.CreateHistogram<double>("messaging.process.duration", unit: "s",
            description: "Duration of processing a message");
        Failures = meter.CreateCounter<long>("bookstore.messaging.failures", unit: "{message}");
    }
}

Demoda bir MeterListener, kaydedilen her ölçümü yazdırıyor:

metric messaging.process.duration=0.0139s messaging.system=rabbitmq messaging.destination.name=lab12.billing outcome=success

messaging.process.duration adı ve saniye birimi OpenTelemetry messaging semantic convention'ından. Bu convention'lar hâlâ "Development" statüsünde ve hata durumu için error.type attribute'unu öneriyor; ben burada okunurluk için outcome kullandım, standarda uymak istersen hata dalında exception tipini error.type olarak ekle. Attribute seçerken kardinaliteye dikkat: queue adı, sonuç, sistem gibi az değerli attribute'lar metric'e; MessageId, OrderId gibi her mesajda değişen değerler yalnız span'e ve log'a girer. Her siparişe ayrı bir zaman serisi açan bir metric, metric backend'ini kısa sürede çökertir.

Bu histogram'dan iki şey okunur: p95/p99 işlem süresi (consumer yavaşlıyor mu) ve saniyedeki işlem sayısı (consumer ne kadar iş yapıyor). Hata sayacını retry ve DLQ (Dead-Letter Queue) sayaçlarıyla birlikte koymak, serinin sekizinci ve dokuzuncu yazılarındaki alarmların temelini oluşturur.

Queue Depth ve Consumer Lag

İşlem süresi consumer'ın içini gösterir; geride biriken işi göstermez. RabbitMQ ve Service Bus'ta bu sayı queue depth'tir, yani bekleyen mesaj sayısı. RabbitMQ'da QueueDeclarePassiveAsync hazır mesaj sayısını verir (bir önceki yazılardaki DLQ health check'i bunu kullanıyordu), Prometheus plugin'i queue bazında metric yayınlar; Service Bus'ta ServiceBusAdministrationClient.GetQueueRuntimePropertiesAsync ActiveMessageCount'u döndürür. Kafka'da mesajlar işlenince silinmez; ölçülen şey consumer lag, yani her partition'da son offset ile grubun commit ettiği offset arasındaki fark:

public static class ConsumerLag
{
    public static async Task<object> GetAsync(string bootstrap, string groupId, string topic)
    {
        using IAdminClient admin = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrap }).Build();
        Metadata metadata = admin.GetMetadata(topic, TimeSpan.FromSeconds(5));
        List<TopicPartition> partitions = [.. metadata.Topics[0].Partitions.Select(p => new TopicPartition(topic, p.PartitionId))];

        // grubun commit ettiği offset'ler
        List<ListConsumerGroupOffsetsResult> committed = await admin.ListConsumerGroupOffsetsAsync(
            [new ConsumerGroupTopicPartitions(groupId, partitions)]);

        // partition'ların son offset'i (high watermark)
        using IConsumer<Ignore, Ignore> probe = new ConsumerBuilder<Ignore, Ignore>(
            new ConsumerConfig { BootstrapServers = bootstrap, GroupId = $"{groupId}-lag-probe" }).Build();
        var perPartition = committed[0].Partitions.Select(tpo =>
        {
            WatermarkOffsets w = probe.QueryWatermarkOffsets(tpo.TopicPartition, TimeSpan.FromSeconds(5));
            long position = tpo.Offset == Offset.Unset ? w.Low.Value : tpo.Offset.Value;
            return new { partition = tpo.Partition.Value, committed = position, end = w.High.Value, lag = w.High.Value - position };
        }).ToList();
        return new { group = groupId, topic, totalLag = perPartition.Sum(p => p.lag), partitions = perPartition };
    }
}

Uygulamayı Notification worker'ı kapalı başlattım, 5 sipariş verdim, sonra worker açık olarak yeniden başlattım. GET /lag/notification'ın üç ölçümü:

{"group":"notification","topic":"bookstore.invoices","totalLag":0,"partitions":[{"partition":0,"committed":65,"end":65,"lag":0},{"partition":1,"committed":81,"end":81,"lag":0},{"partition":2,"committed":73,"end":73,"lag":0}]}
{"group":"notification","topic":"bookstore.invoices","totalLag":5,"partitions":[{"partition":0,"committed":65,"end":67,"lag":2},{"partition":1,"committed":81,"end":82,"lag":1},{"partition":2,"committed":73,"end":75,"lag":2}]}
{"group":"notification","topic":"bookstore.invoices","totalLag":0,"partitions":[{"partition":0,"committed":67,"end":67,"lag":0},{"partition":1,"committed":82,"end":82,"lag":0},{"partition":2,"committed":75,"end":75,"lag":0}]}

Beş fatura üç partition'a dağıldı (key = OrderId) ve worker yokken lag 5'e çıktı; worker geri gelince sıfırlandı. Geride kalan mesajlar işlenirken her biri kendi eski trace'ine bağlandı: Notification'ın process span'i, siparişin dakikalar önce başlamış trace'ine eklendi ve trace görüntüleyicide bu bekleme süresi açıkça görünür. Lag'in mutlak değeri kadar eğilimi de önemli: sabit 50 sağlıklı olabilir, 10 dakikada 0'dan 5000'e çıkan lag consumer'ın geride kaldığını söyler. Lag'i zamana çevirmek (en eski işlenmemiş mesajın yaşı) alarm eşiğini iş diliyle konuşmayı kolaylaştırır: "Notification 2 dakikadan fazla geride".

Üretimde Dikkat

Sampling'i bilinçli seç. Her mesajın trace'ini saklamak pahalıdır. Head-based sampling'de karar trace'in başında verilir ve traceparent'taki sampled bayrağıyla bütün servislere taşınır; HTTP isteği örneklenmediyse mesaj zincirinin tamamı da örneklenmez. Hatalı ya da yavaş trace'leri seçerek saklamak istiyorsan bu karar Collector'da tail-based sampling ile verilir.

Diğer client'lar. Azure.Messaging.ServiceBus'ın ActivitySource desteği deneysel ve Azure.Experimental.EnableActivitySource switch'i ile açılıyor; sonra AddSource("Azure.*") ile dinlenir. MassTransit kendi kaynağını yayınlar ve DiagnosticHeaders.DefaultListenerName ile eklenir. Hangi client hangi header'ı kullanıyor, kontrol et: aynı sistemde iki farklı propagation formatı (ör. eski Diagnostic-Id ile traceparent) varsa zincir ortadan kopar.

Outbox'tan geçen mesajlar. Relay ayrı bir arka plan işi olduğu için trace context'i taşımaz; Activity.Current?.Id'yi outbox satırına bir kolon olarak yaz, relay'de header'a koy. Böylece "sipariş kaydı → relay → consumer" zinciri, relay'in beklettiği süreyle birlikte tek trace'te görünür.

Metric'i queue'dan değil broker'dan da al. Uygulama içinden ölçülen queue depth, uygulama çalışmadığında ölçülmez; tam da en çok ihtiyaç duyduğun anda. Queue depth ve lag'i broker'ın kendi exporter'ından (RabbitMQ Prometheus plugin'i, Azure Monitor, Kafka lag exporter'ları) toplamak bu kör noktayı kapatır.

Sık Düşülen Tuzaklar

AddSource'u unutmak. RabbitMQ.Client 7 instrumentation'ı hazır ama dinleyici yoksa span de header da yok; downstream servisler her mesajda yeni trace başlatır.

Header'ı yanlış encoding'le yazmak. Kafka header'ı byte[]'dır; traceparent'ı UTF-8 yaz ve oku. Farklı dillerde yazılmış servisler varsa aynı formatta anlaştığından emin ol.

Yüksek kardinaliteli metric attribute'ları. MessageId ya da CustomerId'yi metric tag'i yapmak, metric backend'ini zaman serisiyle boğar.

Log scope'una trace id'yi elle eklemek. Logger bunu Activity.Current'tan zaten ekliyor; elle eklemek aynı değeri iki kez yazar ve span değişse bile eski değeri taşıyabilir.

Yalnız uygulama içinden queue depth ölçmek. Consumer çöktüğünde metric de susar; alarm hiç çalmaz.

Küçük Bir Deneme

★☆☆ SpanLineProcessor'a span'in süresini (a.Duration) ekle ve deliver order.placed span'inin send bookstore.invoices span'ini kapsadığını sürelerden doğrula.

★★☆ Bir önceki yazıdaki outbox'a TraceParent kolonu ekle; endpoint'te Activity.Current?.Id'yi yaz, relay'de BasicProperties.Headers["traceparent"]'a koy ve relay'in kendi span'ini bu context'ten başlat. Consumer'ın span'inin artık HTTP isteğinin trace'inde göründüğünü doğrula.

★★★ ConsumerLag'i bir ObservableGauge<long>'a bağla (partition başına bir ölçüm, attribute olarak messaging.consumer.group.name ve partition) ve 15 saniyede bir ölçecek şekilde ayarla. Notification'ı durdurup 100 mesaj üret, gauge'un yükselip worker geri gelince düştüğünü göster.

Artık bir mesajın yolculuğunu görebiliyoruz. Bir sonraki yazıda bu yolculuğu otomatik testlerle güvenceye alıyoruz: handler mantığını broker'sız unit test etmek, Testcontainers ile gerçek RabbitMQ'ya integration test, MassTransit test harness'ı ve v1 mesajını v2 consumer'ın okuyabildiğini kanıtlayan contract testleri.