Kubernetes Üzerindeki .NET Core Uygulamalarının OpenTracing API ile Distributed Tracing İşlemleri

Distributed tracing, microservice architecture’ı olarak tasarladığımız sistem içerisindeki uygulamalarımızın, nerede performans problemi yaşadığını belirleyebilmemiz ve monitor edebilmemiz için harika bir method.

Bir başka değişle, hangi request nereye gidiyor, uçtan uca bir request ne kadar zaman harcıyor gibi sorulara cevap alabilmemiz için implemente etmemiz gereken bir method.

Bu makale kapsamında ise, OpenTracing API‘ını ve Jaeger tracer’ını kullanarak .NET Core ile geliştirdiğimiz kubernetes üzerindeki microservice’lerin, distributed tracing işlemlerine değineceğiz.

Senaryo

Bir e-ticaret sistemi üzerinde çalıştığımızı ve uygulamalarımızı kubernetes üzerinde host ettiğimizi düşünelim. Kullanıcılar ile ilgili işlemlerden sorumlu bir User API‘ımız var. Yeni bir kullanıcı sisteme kayıt olduğunda ise, “UserRegisteredEvent” adında bir event publish ediliyor. Publish edilen bu event’in subscriber’larından birisi ise, kullanıcıya hesabını aktifleştirebilmesi için e-posta göndermekle sorumlu bir service.

Biz ise asenkron olarak gerçekleşen bu kullanıcı kayıt journey’inin, OpenTracing API ve Jaeger tracer’ını implemente ederek, kubernetes üzerinde uçtan uca izleme işlemini gerçekleştireceğiz.

OpenTracing ve Jaeger Nedir?

Kısaca sizlere OpenTracing ve Jaeger hakkında bilgi vermek istiyorum. OpenTracing, herhangi bir vendor’a bağımlı olmadan uygulamalarımıza distributed tracing için bir instrumentation ekleyebilmemizi sağlayan bir specification’dır. Tıpkı, OpenAPI gibi.

Jaeger ise Uber Technologies tarafından geliştirilmiş, OpenTracing standart’larını destekleyen ve microservice mimarimiz üzerinde distributed tracing işlemlerini yapabilmemizi sağlayan harika bir tracer’dır. Jaeger hakkında daha detaylı bilgiye ise, buradan ulaşabilirsiniz.

Jaeger‘in kubernetes üzerine kurulumu için, şuradaki dokümanları takip edebilirsiniz. Ben bu makale için, development setup’ını uyguladım.

kubectl create -f https://raw.githubusercontent.com/jaegertracing/jaeger-kubernetes/master/all-in-one/jaeger-all-in-one-template.yml

Jaeger genel hatlarıyla, “Agent“, “Collector” ve “Query” den oluşmaktadır. Agent, UDP üzerinden kendisine gelen trace verilerini dinleyip, collector’e ileten bir network daemon’ıdır. Trace verileri ise “Span” olarak adlandırılmaktadır. Collector ise kendisine iletilen trace verilerini, bir pipeline (validations, indexes, transformations) içerisinde işlemektedir. Daha sonra ise seçtiğiniz bir component (Elasticsearch, Cassandra ve Kafka) türüne göre store etmektedir.

Query ise isminden de anlaşılabileceği üzere, ilgili trace sonuçlarını sorgulayabileceğimiz bir UI.

Peki, Haydi Biraz Kodlayalım!

Kodlamaya başlamadan önce platform için sahip olmamız gereken bazı tool’lar:

• Message Broker (ben RabbitMQ kullanacağım)
• Docker
• ve Kuberentes

NOT: Ana konumuz platformu oluşturmak olmadığı için, ben kurulum konuları üzerinde durmayacağım.

İlk olarak kullanıcıların sisteme kayıt olabilmelerini sağlayacak olan User API‘ını develop edelim. Bunun için öncelikle aşağıdaki gibi bir proje oluşturalım.

dotnet new webapi -n User.API

Ardından “User.Common.Contracts” isminde bir class library oluşturalım.

dotnet new classlib -n User.Common.Contracts

Bu library içerisinde uygulamalarımız arasında share edeceğimiz contract’ları tanımlayacağız. Şimdi yeni bir kullanıcı sisteme kayıt olduğunda publish edeceğimiz event’i, aşağıdaki gibi oluşturalım.

using System.Collections.Generic;

namespace User.Common.Contracts
{
    public class UserRegisteredEvent
    {
        public string Email { get; set; }
        public Dictionary<string, string> TracingKeys { get; set; }
    }
}

Oluşturma işleminin ardından, “User.Common.Contracts” library’sini, az önce oluşturmuş olduğumuz “User.API” projesine referans olarak ekleyelim.

dotnet add reference ../User.Common.Contracts/User.Common.Contracts.csproj

Şimdi “User.API” projesi içerisinde “Models” isminde bir klasör oluşturalım ve ardından içerisine “Requests” ve “Responses” klasörlerini de oluşturalım.

“Requests” klasörü içerisine kullanıcının kayıt olurken kullanacağı modeli aşağıdaki gibi tanımlayalım.

namespace User.API.Models.Requests
{
    public class CreateUserRequest
    {
        public string Username { get; set; }
        public string Password { get; set; }
        public string Email { get; set; }
    }
}

“Responses” klasörü içerisine ise, internal response wrapper class’ını tanımlayalım.

using System.Collections.Generic;
using System.Linq;

namespace User.API.Models.Responses
{
    public class BaseResponse
    {
        public BaseResponse()
        {
            Errors = new List();
        }

        public T Data { get; set; }
        public List Errors { get; set; }
        public bool HasError { get { return Errors.Any();  } }
    }
}

Şimdi ise “Services” isminde bir klasör oluşturalım “User.API” projesi içerisinde. Ardından içerisine aşağıdaki gibi “IUserService” isminde bir interface tanımlayalım.

using System.Threading.Tasks;
using User.API.Models.Requests;
using User.API.Models.Responses;

namespace User.API.Services
{
    public interface IUserService
    {
         Task CreateUserAsync(CreateUserRequest request);
    }
}

Kullanıcı ile ilgili business logic’leri, bu service aracılığı ile gerçekleştireceğiz.

Bu noktada messaging ile ilgili işlemleri reliable bir şekilde gerçekleştirebilmemiz için projemize NuGet üzerinden bir service bus ekleyeceğiz. Ben MassTransit‘in lightweight bir wrapper’ı olan MetroBus library’sini kullanacağım.

dotnet add package MetroBus
dotnet add package MetroBus.Microsoft.Extensions.DependencyInjection

Ardından distributed tracing işlemlerini gerçekleştirebilmemiz için ise OpenTracing ve Jaeger package’larını eklememiz gerekmektedir.

dotnet add package OpenTracing.Contrib.NetCore
dotnet add package Jaeger

Şimdi bir klasöre daha ihtiyacımız var. Ben structured klasör yapılarını seviyorum. Her neyse, “Services” klasörü altında, “Implementations” adında bir klasör oluşturalım ve içerisinde “IUserService” interface’ini aşağıdaki gibi implemente edelim.

using System;
using System.Threading.Tasks;
using User.API.Models.Requests;
using User.API.Models.Responses;
using MassTransit;
using Microsoft.Extensions.Logging;
using User.Common.Contracts;
using OpenTracing;
using OpenTracing.Tag;
using OpenTracing.Propagation;
using System.Collections.Generic;

namespace User.API.Services.Implementations
{
    public class UserService : IUserService
    {
        private readonly ILogger _logger;
        private readonly IBusControl _busControl;
        private readonly ITracer _tracer;

        public UserService(ILogger logger, IBusControl busControl, ITracer tracer)
        {
            _logger = logger;
            _busControl = busControl;
            _tracer = tracer;
        }

        public async Task CreateUserAsync(CreateUserRequest request)
        {
            BaseResponse createUserResponse = new BaseResponse();

            try
            {
                using (var scope = _tracer.BuildSpan("create-user-async").StartActive(finishSpanOnDispose: true))
                {
                    var span = scope.Span.SetTag(Tags.SpanKind, Tags.SpanKindClient);

                    var dictionary = new Dictionary<string, string>();
                    _tracer.Inject(span.Context, BuiltinFormats.TextMap, new TextMapInjectAdapter(dictionary));

                    //some user create business logics

                    createUserResponse.Data = 1; // User id

                    await _busControl.Publish(new UserRegisteredEvent
                    {
                        Email = request.Email,
                        TracingKeys = dictionary
                    });
                }
            }
            catch (Exception ex)
            {
                createUserResponse.Errors.Add(ex.Message);
                _logger.LogError(ex, ex.Message);
            }

            return createUserResponse;
        }
    }
}

Service içerisinde kısaca neler yaptık bir bakalım.

Trace context’inin diğer service’lere otomatik olarak propagate edilmesi Jaeger tarafından zaten otomatik olarak sağlanıyor. Yani api-to-api communication’ı olduğunda ve gerekli configuration’ı yaptığınızda, bir request’i uçtan uca izleyebiliyorsunuz.

Biz bu noktada ise api-to-subscriber olarak event-based bir communication gerçekleştirdiğimiz için, trace context’inin propagation işlemini manuel olarak kendimiz sağladık. Trace scope’u içerisine bakarsak, “create-user-async” isminde bir span oluşturduk. Sonrasında ise client olduğuna dair bir tag ekledik. Tag’leri kullanarak span’a additional metadata’lar ekleyebilmek mümkündür. Ardından ise “TextMapInjectAdapter” aracılığı ile ilgili trace context’ini dictionary’e inject ettik.

Kullanıcının sisteme kayıt olabilme işlemlerini de tamamlayarak, tracing key’leri ile birlikte “UserRegisteredEvent” ini service bus aracılığı ile queue’ya publish ettik. Bu noktadan sonra ilgili event’i her kim consume ederse, tracing key’lerini kullandığı sürece tüm akışı gözlemleyebileceğiz.

Artık controller’ı oluşturabiliriz. Aşağıdaki gibi “UsersController” isminde bir controller oluşturalım.

using Microsoft.AspNetCore.Mvc;
using User.API.Services;
using User.API.Models.Requests;
using User.API.Models.Responses;
using System.Threading.Tasks;

namespace User.API.Controllers
{
    [Route("api/users")]
    [ApiController]
    public class UsersController : ControllerBase
    {
        private readonly IUserService _userService;

        public UsersController(IUserService userService)
        {
            _userService = userService;
        }

        [HttpPost]
        public async Task Post([FromBody]CreateUserRequest request)
        {
            BaseResponse createUserResponse = await _userService.CreateUserAsync(request);

            if(!createUserResponse.HasError)
            {
                return Created("users", createUserResponse.Data);
            }
            else
            {
                return BadRequest(createUserResponse.Errors);
            }
        }
    }
}

Controller içerisinde ise, oluşturduğumuz “IUserService” interface’i üzerinden kullanıcının kayıt işlemlerini gerçekleştiriyoruz.

Şimdi “Startup” class’ını açalım ve ilgili service injection işlemlerini aşağıdaki gibi gerçekleştirelim.

public void ConfigureServices(IServiceCollection services)
{
    services.AddMvc().SetCompatibilityVersion(CompatibilityVersion.Version_2_2);

    services.AddScoped<IUserService, UserService>();

    string rabbitMqUri = Configuration.GetValue("RabbitMqUri");
    string rabbitMqUserName = Configuration.GetValue("RabbitMqUserName");
    string rabbitMqPassword = Configuration.GetValue("RabbitMqPassword");

    services.AddSingleton(MetroBusInitializer.Instance.UseRabbitMq(rabbitMqUri, rabbitMqUserName, rabbitMqPassword).Build());

    services.AddOpenTracing();
    services.AddSingleton(serviceProvider =>
    {
        Environment.SetEnvironmentVariable("JAEGER_SERVICE_NAME", "User.API");
        Environment.SetEnvironmentVariable("JAEGER_AGENT_HOST", "localhost");
        Environment.SetEnvironmentVariable("JAEGER_AGENT_PORT", "6831");
        Environment.SetEnvironmentVariable("JAEGER_SAMPLER_TYPE", "const");
        
        var loggerFactory = new LoggerFactory();

        var config = Jaeger.Configuration.FromEnv(loggerFactory);
        var tracer = config.GetTracer();

        GlobalTracer.Register(tracer);

        return tracer;
    });

    services.AddHealthChecks();
}

public void Configure(IApplicationBuilder app, IHostingEnvironment env)
{
    //...
    
    app.UseHealthChecks("/health");
}

Service bus’ı, RabbitMQ kullanarak initialize ettik ve ardından injection işlemini gerçekleştirdik. Daha sonra ise tracer’ı configure ederek inject ettik. Tracer’ı yapılandırırken ben sampling type’ı olarak, “const” sampler’ı kullandım. Diğer seçebileceğiniz sampling seçenekleri ise “Probabilistic“, “Rate Limiting” ve “Remote” şeklinde. Daha detaylı sampling bilgilerine ise buradan erişebilirsiniz. Agent host bilgisini de kubernetes ortamınızdaki node IP‘si ile değiştirebilirsiniz. Eğer agent kurulumunu sidecar olarak gerçekleştirirseniz de, herhangi bir bilgi set etmenize gerek olmayacaktır. Default bilgilerle erişim sağlayacaktır.

API artık hazır durumda. Senaryomuza tekrar dönelim. Kullanıcı sisteme kayıt olduktan sonra bir event publish edecektik. Daha sonra ise kullanıcının hesabını aktifleştirebilmesi için o event’e subscribe olmuş bir aktivasyon e-posta’sı gönderen service oluşturacaktık.

Şimdi event’i publish ettik ve artık kullanıcıya aktivasyon e-posta’sını gönderecek olan service’i kodlamaya başlayabiliriz.

Subscriber’ın Kodlanması

Bunun için yeni bir .NET Core console application’ı oluşturalım.

dotnet new console -n User.Activation.Consumer

Oluşturmanın ardından shared library olan “User.Common.Contracts” projesini referans olarak ekleyelim. Daha sonra ise NuGet üzerinden MetroBus‘ı, OpenTracing‘i ve Jaeger’i projeye aşağıdaki gibi dahil edelim.

dotnet add package MetroBus
dotnet add package MetroBus.Microsoft.Extensions.DependencyInjection
dotnet add package OpenTracing.Contrib.NetCore 
dotnet add package Jaeger

Console uygulaması, deamon olarak çalışacak bir background service olacak. App startup ve lifetime management’ını yapabilmemiz için ise NuGet üzerinden “Microsoft.Extensions.Hosting” ve “Microsoft.Extensions.DependencyInjection” paketlerini de projeye dahil edelim.

dotnet add package Microsoft.Extensions.Hosting
dotnet add package Microsoft.Extensions.DependencyInjection

dotnet add package Microsoft.Extensions.Configuration
dotnet add package Microsoft.Extensions.Configuration.Json

using System;
using System.Collections.Generic;
using OpenTracing;
using OpenTracing.Propagation;
using OpenTracing.Tag;

namespace User.Activation.Consumer.Common
{
    public static class TracingExtension
    {
        public static IScope StartServerSpan(ITracer tracer, IDictionary<string, string> headers, string operationName)
        {
            ISpanBuilder spanBuilder;
            try
            {
                ISpanContext parentSpanCtx = tracer.Extract(BuiltinFormats.TextMap, new TextMapExtractAdapter(headers));

                spanBuilder = tracer.BuildSpan(operationName);
                if (parentSpanCtx != null)
                {
                    spanBuilder = spanBuilder.AsChildOf(parentSpanCtx);
                }
            }
            catch (Exception)
            {
                spanBuilder = tracer.BuildSpan(operationName);
            }

            return spanBuilder.WithTag(Tags.SpanKind, Tags.SpanKindConsumer).StartActive(true);
        }
    }
}

using System.Threading.Tasks;
using MassTransit;
using OpenTracing;
using User.Activation.Consumer.Common;
using User.Common.Contracts;

namespace User.Activation.Consumer.Consumers
{
    public class UserActivationConsumer : IConsumer<UserRegisteredEvent>
    {
        private readonly ITracer _tracer;

        public UserActivationConsumer(ITracer tracer)
        {
            _tracer = tracer;
        }

        public async Task Consume(ConsumeContext<UserRegisteredEvent> context)
        {
            using (var scope = TracingExtension.StartServerSpan(_tracer, context.Message.TracingKeys, "user-activation-link-sender-consumer"))
            {
                //some user activation link send business logics

                await System.Console.Out.WriteLineAsync($"Activation link sent for {context.Message.Email}");
            }
        }
    }
}

using System.Threading;
using System.Threading.Tasks;
using MassTransit;
using Microsoft.Extensions.Hosting;

namespace User.Activation.Consumer.Services.Implementations
{
    public class BusService : IHostedService
    {
        private readonly IBusControl _busControl;

        public BusService(IBusControl busControl)
        {
            _busControl = busControl;
        }

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

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

using System;
using System.IO;
using System.Threading.Tasks;
using Jaeger;
using Jaeger.Samplers;
using MassTransit;
using MetroBus;
using MetroBus.Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using OpenTracing;
using OpenTracing.Util;
using User.Activation.Consumer.Consumers;
using User.Activation.Consumer.Services.Implementations;

namespace User.Activation.Consumer
{
    class Program
    {
        static async Task Main(string[] args)
        {
            var host = new HostBuilder()
                .ConfigureAppConfiguration((hostingContext, config) =>
                {
                    config.SetBasePath(basePath: Directory.GetCurrentDirectory());
                    config.AddJsonFile("appsettings.json", optional : true);
                })
                .ConfigureServices((hostContext, services) =>
                {
                    //Init tracer
                    services.AddSingleton<ITracer>(t => InitTracer());

                    string rabbitMqUri = hostContext.Configuration.GetValue<string>("RabbitMqUri");
                    string rabbitMqUserName = hostContext.Configuration.GetValue<string>("RabbitMqUserName");
                    string rabbitMqPassword = hostContext.Configuration.GetValue<string>("RabbitMqPassword");

                    services.AddMetroBus(x =>
                    {
                        x.AddConsumer<UserActivationConsumer>();
                    });

                    services.AddSingleton<IBusControl>(provider => MetroBusInitializer.Instance
                        .UseRabbitMq(rabbitMqUri, rabbitMqUserName, rabbitMqPassword)
                        .RegisterConsumer<UserActivationConsumer>("user.activation.queue", provider)
                        .Build());

                    services.AddHostedService<BusService>();
                });

            await host.RunConsoleAsync();
        }

        private static ITracer InitTracer()
        {
            Environment.SetEnvironmentVariable("JAEGER_SERVICE_NAME", "User.Activation.Consumer");
            Environment.SetEnvironmentVariable("JAEGER_AGENT_HOST", "localhost");
            Environment.SetEnvironmentVariable("JAEGER_AGENT_PORT", "6831");
            Environment.SetEnvironmentVariable("JAEGER_SAMPLER_TYPE", "const");

            var loggerFactory = new LoggerFactory();

            var config = Jaeger.Configuration.FromEnv(loggerFactory);
            var tracer = config.GetTracer();

            GlobalTracer.Register(tracer);

            return tracer;
        }
    }
}

Sanırım yukarıda yaptıklarımız yeterince açık ve net. Configuration’ı ve dependency injection’ı configure ederek, service’lerimizi inject ediyoruz. Ayrıca trace’i ise, “const” type’ı ile ve “User.Activation.Consumer” adıyla initialize ediyoruz.

Consumer’ı ise, “user.activation.queue” adında bir queue ile register ediyoruz. Bu queue ile “UserRegisteredEvent” model’ine subscribe olacaktır.

Deployment

Artık uygulamaları deploy etmeye hazırız. Ben deployment işlemlerini gerçekleştirebilmek için basit bir Docker file ve Helm chart hazırladım. Bu chart ile, uygulamaları Azure Kubernetes Service üzerine deploy edeceğim. Siz kendi environment’ınıza göre chart’ı değiştirebilirsiniz.

Hazırlamış olduğum chart ve docker file’a, buradan erişebilirsiniz.

Test

Şimdi test aşamasına geçebiliriz. Öncelikle sistemde yeni bir kullanıcı oluşturabilmek için aşağıda olduğu gibi “api/users” endpoint’ine bir POST request’i gönderelim.

Bu request ile kullanıcı kayıt journey’ini başlatmış olduk. Senaryomuzda olduğu gibi, kullanıcı kayıt işlemi gerçekleştikten sonra bir event publish edildi.

Event’in publish edilmenin ardından event’e subscribe etmiş olduğumuz aktivasyon e-posta’sını gönderecek olan service (user-activation-consumer) ilgili işlemini gerçekleştirmiştir.

Peki bu süreçte neler oldu, haydi Jaeger üzerinden bir bakalım.

Jaeger üzerindeki akışa bakarsak, bu işlem 4 derinliğe ve 5 adet span’a sahip. Toplam süreç ise 29.54ms sürmüş. Post işleminin kırılımlara bakarsak ise, request ilgili action’dan geçtikten sonra “create-user-async” method’una geliyor ve içerisinde kullanıcı kayıt işlemleri gerçekleştiriliyor. Ardından ise asenkron olarak aktivasyon link’i gönderme işlemleri “User.Activation.Consumer” service’i içerisinde gerçekleştiriliyor.

Gördüğümüz gibi bu journey asenkron olarak gerçekleşiyor olmasına rağmen, bu request nerede, süreç nerede ne kadar zaman harcıyor gibi sorulara cevap alabiliyoruz.

Sonuç

Distributed tracing ile bir developer olarak microservice mimarisi içerisinde koşturan kodumuzu, request’in life cycle’ını, debug edebilir ve optimize edebiliriz. OpenTracing API’ı ile de, vendor lock-in durumuna düşmeden sistemimizin farklı tracer’lar ile esnek bir şekilde trace edilebilmesini sağlayabiliyoruz. Ayrıca ben bu makalede, distributed bir yapıdaki sistem içerisinde trace bilgilerinin propagation işlemlerini de göstermeye çalıştım.

Projects: https://github.com/GokGokalp/OpenTracing-Jaeger-Sample

Referanslar

https://github.com/yurishkuro/opentracing-tutorial
https://www.jaegertracing.io/docs/1.11/architecture/