Aspire İle .NET’te Dapr Workflows Oluşturma

Merhaba,

İçerik sürecinde Dapr workflow’lardan bahsederken olabilecek bir karışıklılığa temas etmekte fayda görmekteyim. O da, Dapr’ın kendisinin bir workflow motoru olmadığı, Dapr’ın sunduğu building block’lardan birinin “workflow” olduğudur.

Dapr Workflow Gerçekte Nedir?

İlk bakışta Dapr workflows, belirli adımları sırayla çalıştıran bir iş akışı mekanizması gibi görünebilir. Ancak gerçekte sunduğu şey tabi ki de bundan çok daha fazlasıdır. Dapr workflows; uzun süren (long-running), dayanıklı (durable) ve hata toleranslı (fault-tolerant) iş süreçlerini yönetebilmek için geliştirilmiş bir workflow orchestration altyapısıdır. Bir başka ifadeyle, bir iş sürecinin yalnızca hangi adımlardan oluşacağını değil; bu adımların güvenli şekilde yürütülmesini, başarısızlık durumunda tekrar devam edebilmesini ve tüm sürecin durumunun korunmasını sağlayan sorumlu bir işlevselliktir.

Örneğin yandaki gibi bir sipariş sürecini ele alalım… İlk bakışta bu işlemler sıradan metot çağrıları gibi görünebilir. Ancak biliyorsunuz ki gerçek hayatta işler çoğu zaman bu kadar sorunsuz ilerlememektedir. Ödeme servisi geçici olarak erişilemeyebilir, stok servisi cevap vermeyebilir, kargo sistemi birkaç saat sonra kullanılabilir hale gelebilir, uygulama tam sürecin ortasında yeniden başlatılabilir, sunucu çökebilir yahut yeni bir sunucuya taşınabilir…

Klasik bir uygulamada bu senaryoların tamamını geliştiricinin yönetmesi gerekmektedir. Sürecin hangi adımda kaldığını saklamak, yeniden denemeleri planlamak, başarısız olan işlemleri telafi etmek (compensation) ve akışı doğru noktadan devam ettirmek çoğu zaman ciddi bir mühendislik yükü oluşturmaktadır.

İşte Dapr workflows tam olarak bu noktada devreye girmektedir. Workflow’un mevcut durumunu kalıcı olarak saklar, uygulama yeniden başlasa bile kaldığı yerden devam edebilir, zamanlayıcılar oluşturabilir, dış olayları bekleyebilir ve tüm iş akışını tek bir orchestration altında yönetebilir. Bu sayede geliştirici, altyapısal ayrıntılar yerine yalnızca “iş süreci nasıl ilerlemeli?” sorusuna odaklanabilir…

Dapr workflows, metotları sırasıyla çalıştıran bir kütüphane değildir; iş süreçlerini güvenilir bir şekilde yürüten ve durumunu koruyan bir orchestarion motorudur.

Durable Execution Nedir?

Dapr workflows’un temeli esasında kalıcı yürütme dediğimiz durable execution kavramına dayanmaktadır. Ve hatta Dapr workflows’u anlamanın en doğru yolu önce bunu anlamaktır.

En basit tanımıyla durable execution, çalışan bir iş akışının durumunun kalıcı olarak saklanması ve herhangi bir kesinti yaşansa bile kaldığı yerden devam edebilmesidir.

Bunu daha iyi anlayabilmek için aşağıdaki gibi bir akış üzerinden karşılaştırmada bulunabiliriz;

Şöyle bir kodu çalıştırdığımızı varsayalım:

await TakePayment();
await UpdateStock();
await CreateInvoice();
await SendEmail();

Şimdi düşünelim…

Fatura oluşturma sürecinde sunucunun çöktüğünü varsayalım… Ee ne olacak? Hiçbir şey… Sunucu yeniden ayağa kalkacak… Ancak CLR belleği sıfırlanacağı, stack boşaltılacağı ve local değişkenler gideceği için workflow’un hangi adımda olduğu bilgisi de gitmiş olacaktır. Yani bu akışta süreç tekrar başlatılırsa muhtemelen TakePayment() metodu yeniden çalıştırılacaktır. Bu da ikinci kez para çekilmesi gibi felaketlere sebep olacaktır 😱

Durable execution ile ise workflow her activity tamamlandığında durumunu kalıcı storage’a yazacaktır.

Böylece sunucu kapansa da, Docker silinse de, Kubernetes Pod’u yeniden oluşturulsa da yahut makine yeniden başlasa da hiç fark etmeyecek! workflow tekrar ayağa kalktığında storage’dan kaldığı yerden devam edecektir.

Durable Execution, esasında ‘kodun hafızasıdır’…

Peki Dapr bunu nasıl yapıyor?
Workflow gerçekte kaldığı satırdan devam etmiyor. Arka planda event sourcing mantığına benzer bir yaklaşım kullanıyor. Her activity tamamlandığında geçmiş (history) kaydediliyor. Workflow yeniden çalıştırılırken orchestration kodu deterministik olarak tekrar yürütülüyor, ancak tamamlanmış activity’ler yeniden çalıştırılmıyor. Bunun yerine daha önce kaydedilmiş sonuçlar kullanılıyor. Kod, sanki hiç kesinti yaşanmamış gibi kaldığı noktaya ulaşıyor ve yalnızca sıradaki activity gerçek anlamda çalıştırılıyor.

Bundan kaynaklı Dapr workflows’ta orchestration kodunun deterministik olması elzemdir. Rastgele sayı üretmek, DateTime.Now okumak yahut dış servislere doğrudan istekte bulunmak gibi işlemler orchestration içinde önerilmemektedir. Bu tarz çalışmalar activity’lerde gerçekleştirilmelidir.

Dapr Workflows Temel Kavramları

Dapr workflows’u anlamanın en kolay yolunun, onu oluşturan temel kavramlara yani terminolojiye hakim olmaktan geçtiğini düşünüyorum.

Aspire’da Dapr Workflow’un Oluşturulması

Şimdi Aspire üzerinden Dapr workflow’un nasıl oluşturulduğunu hep beraber incelemeye başlayalım… Tabi öncelikle akıllara neden Aspire kullanıyoruz? sorusunun geldiğini görür gibiyim… Hemen bu suali cevaplandırarak devam edelim… Geliştireceğimiz uygulamada Aspire ile Dapr sidecar’ı ve state store’u uygulamayla birlikte tek bir noktadan ayağa kaldırabilmekte ve OpenTelemetry tabanlı gözlemlenebilirlik (observability) desteği sayesinde logları, metrikleri ve trace’leri de merkezi olarak takip edebilme olanağı elde edebilmekteyiz. Böylece altyapı konfigürasyonlarıyla uğraşmak yerine doğrudan Dapr workflows’u geliştirmeye ve davranışlarını gözlemlemeye odaklanabileceğiz.

Evet… Artık fiziksel olarak Dapr workflow’u tatbik etmeye gelirsek eğer tabi ki de öncelikle bir Aspire uygulaması oluşturulmalı ve bir yandan da içerisine, Dapr’ın sidecar olarak eşlik edeceği bir Asp.NET Core Web API projesi eklenmelidir.

Burada API projesine Dapr.Workflow, Aspire’a ise CommunityToolkit.Aspire.Hosting.Dapr paketlerinin kurulması gerekmektedir.

Şimdi aşağıdaki adımları sırasıyla seyrederek Dapr workflow’u oluşturmaya geçelim…

• Adım 1 (Activity’lerin Oluşturulması)
  İlk olarak, workflow’da yapılacak tüm iş mantıklarının somut olarak tanımlanmasını gerçekleştirelim. Bunun için aşağıdaki tanımlarda olduğu gibi WorkflowActivity<TInput, TOutput> abstract class’ını implemente eden sınıflar oluşturmalıyız.
  • CheckInventoryActivity : Envanter kontrolü gerçekleştirmektedir.

        public sealed class CheckInventoryActivity(IInventoryService inventoryService) : WorkflowActivity<OrderPayload, InventoryResult>
        {
            public override async Task<InventoryResult> RunAsync(WorkflowActivityContext context, OrderPayload input)
            {
                bool inStock = await inventoryService.HasStockAsync(input.ProductId, input.Quantity);
    
                return new InventoryResult(inStock, inStock ? "Stok mevcuttur." : "Stok yetersiz.");
            }
        }
    

  • ProcessPaymentActivity : Sipariş ödemesini gerçekleştirmektedir.

        public sealed class ProcessPaymentActivity : WorkflowActivity<PaymentRequest, object?>
        {
            public override Task<object?> RunAsync(WorkflowActivityContext context, PaymentRequest input)
            {
                Console.WriteLine($"{{{input.OrderId}}} siparişi için {{{input.Amount:C}}} tutarında ücret tahsil edilmektedir.");
    
                return Task.FromResult<object?>(null);
            }
        }
    

  • UpdateInventoryActivity : Stok güncellemektedir.

        public sealed class UpdateInventoryActivity(IInventoryService inventoryService) : WorkflowActivity<OrderPayload, object?>
        {
            public override async Task<object?> RunAsync(WorkflowActivityContext context, OrderPayload input)
            {
                await inventoryService.ReserveStockAsync(input.ProductId, input.Quantity);
    
                Console.WriteLine($"{{{input.OrderId}}} siparişi için {{{input.Quantity}}} adet ürün stokta ayrılmıştır.");
    
                return null;
            }
        }
    

  • SendMailActivity : Müşteriye mail ile bilgilendirme göndermektedir.

        public sealed class SendMailActivity : WorkflowActivity<string, object?>
        {
            public override Task<object?> RunAsync(WorkflowActivityContext context, string input)
            {
                Console.WriteLine($"Müşteriye e-posta gönderilmiştir: '{input}'");
                return Task.FromResult<object?>(null);
            }
        }
    

  • NotifyCustomerActivity : Müşteriye bildiri göndermektedir.

        public sealed class NotifyCustomerActivity : WorkflowActivity<string, object?>
        {
            public override Task<object?> RunAsync(WorkflowActivityContext context, string input)
            {
                Console.WriteLine($"Müşteriye {{{input}}} bildirimi gönderilmiştir.");
    
                return Task.FromResult<object?>(null);
            }
        }
    

• Adım 2 (Workflow’un Tasarlanması)
  Evet… Artık activity’ler hazır olduğuna göre workflow’u tasarlayabiliriz. Bunun için de
  Workflow<TInput, TOutput> abstract class’ından istifade ediyor olacağız.

      public class OrderProcessingWorkflow : Workflow<OrderRequest, OrderResult>
      {
          public override async Task<OrderResult> RunAsync(WorkflowContext context, OrderRequest input)
          {
              // 1. Envanter kontrolü yapılır.
              var inventory = await context.CallActivityAsync<InventoryResult>(nameof(CheckInventoryActivity), input);
  
              if (!inventory.Success)
                  return new OrderResult(input.OrderId, false, inventory.Message);
  
              // 2. Müşteriden ücret alınır.
              await context.CallActivityAsync(nameof(ProcessPaymentActivity), new PaymentRequest(input.OrderId, input.TotalAmount));
  
              // 3. Stok güncellenir.
              await context.CallActivityAsync(nameof(UpdateInventoryActivity), input);
  
              // 4. Müşteriye e-posta gönderilir.
              await context.CallActivityAsync(nameof(SendMailActivity), $"{{{input.OrderId}}} numaralı sipariş başarıyla işlenmiştir.");
  
              // 5. Müşteri bilgilendirilir.
              await context.CallActivityAsync(nameof(NotifyCustomerActivity), "Siparişiniz başarıyla işlenmiştir.");
  
              return new OrderResult(input.OrderId, true, "Sipariş başarıyla işlenmiştir.");
          }
      }
  

  Burada görüldüğü üzere iş süreci uçtan uca tanımlanmaktadır. Bu tanımlama sürecinde sürekli kullanılan CallActivityAsync metodu ise belirtilen activity’nin çalıştırılması için bir task planlamaktadır. Bu task tamamlanınca sonucu almakta ve workflow’un durumunu (history/state) kalıcı olarak saklamaktadır. Ve gerektiği taktirde workflow yeniden oynatılarak (replay) bir sonraki adımdan devam edilmektedir.

  Ayrıca şunu bilmekte fayda var ki, burada yaptığımız davranış esasında Task Chaining (Görev Zincirleme) pattern’ına karşılık gelmektedir. Yani workflow içerisindeki her activity, kendisinden önceki activity’nin tamamlanmasını beklemekte ve böylece süreç adım adım ilerlemektedir.

  Task Chaining, Dapr workflows’un temel workflow pattern’ıdır.

  Yapısı gereği workflow doğrusal (linear) bir akış izlemektedir. Dolayısıyla bir sonraki adıma geçilebilmesi için mevcut adımın başarıyla tamamlanması gerekmektedir. Bu sayede iş süreci belirlenen sıraya uygun olarak ilerlemekte ve adımlar arasında tutarlı bir yürütme sağlanmaktadır. Ancak Dapr workflows yalnızca sıralı iş akışları oluşturmakla da sınırlı değildir! Bunun yanında çok daha gelişmiş workflow desenlerini de tabi ki de desteklemektedir. Misal olarak; Fan-out / Fan-in pattern’ında, workflow belirli bir noktada birden fazla activity’yi aynı anda (yani paralel bir şekilde) çalıştırabilmektedir. Bu activity’lerin tamamı sonuçlandığında ise elde edilen çıktılar tek bir noktada birleştirilerek iş akışına devam edilebilmektedir. .NET tarafında bu yaklaşım oldukça doğal bir şekilde uygulanabilmektedir…

• Adım 3 (API Yapılandırmasının Sağlanması)
  Şimdi yaptığımız tüm bu çalışmaları API uygulamasında yapılandıralım. Bunun için ‘Program.cs’ dosyasında aşağıdaki çalışmaları gerçekleştirelim;

  var builder = WebApplication.CreateBuilder(args);
  
  .
  .
  .
  
  builder.Services.AddSingleton<IInventoryService, InventoryService>();
  
  builder.Services.AddDaprWorkflow(configure =>
  {
      configure.RegisterWorkflow<OrderProcessingWorkflow>();
  
      configure.RegisterActivity<CheckInventoryActivity>();
      configure.RegisterActivity<ProcessPaymentActivity>();
      configure.RegisterActivity<UpdateInventoryActivity>();
      configure.RegisterActivity<SendMailActivity>();
      configure.RegisterActivity<NotifyCustomerActivity>();
  });
  
  var app = builder.Build();
  
  .
  .
  .
  
  app.Run();
  

• Adım 4 (Endpoint’lerin Oluşturulması)
  Şimdi de workflow’u çalıştıracak ve bir yandan da workflow’un durumunu kontrol etmemizi sağlayacak endpoint’leri oluşturalım.

  Dapr workflow’u başlatacak endpoint:

  app.MapPost("/orders", async ([FromBody] OrderRequest orderPayload, DaprWorkflowClient daprWorkflowClient) =>
  {
      string instanceId = await daprWorkflowClient.ScheduleNewWorkflowAsync(
          name: nameof(OrderProcessingWorkflow),
          instanceId: orderPayload.OrderId,
          input: orderPayload);
  
      return Results.Accepted($"/orders/{instanceId}", new { instanceId });
  });
  

  Burada kullanılan ScheduleNewWorkflowAsync metodu, yeni bir workflow instance’ını başlatacak ve instance ID değerini döndürecektir. Bu instance ID, workflow’un durumunu sorgulamak için bir sonraki oluşturacağımız endpoint’te kullanılacaktır.

  Belirtilen workflow’un durumunu kontrol edecek endpoint:

  app.MapGet("/orders/{instanceId}", async ([FromRoute] string instanceId, [FromServices] DaprWorkflowClient daprWorkflowClient) =>
  {
      WorkflowState? state = await daprWorkflowClient.GetWorkflowStateAsync(instanceId);
  
      if (state is null || !state.Exists)
          return Results.NotFound();
  
      return Results.Ok(new
      {
          RuntimeStatus = state.RuntimeStatus.ToString(),
          Output = state.ReadOutputAs<OrderResult>()
      });
  });
  

  Bu endpoint’te de dikkat ederseniz GetWorkflowStateAsync metodu kullanılmaktadır. Bu metot, belirtilen instance ID‘ye sahip workflow’un durumunu (state) döndürecektir. WorkflowState referansı ise, workflow’un mevcut durumunu, runtime status’ünü ve output’unu içermektedir. Amacı, workflow’un durumunu sorgulamak ve kullanıcıya geri bildirim sağlamaktır. Eğer workflow instance’ı bulunmazsa, null dönecek ve NotFount() sonucu response edilecektir. RuntimeStatus property’si ise workflow’un mevcut durumunu göstermektedir ve workflow tamamlandığında veya hata aldığında içeriksel olarak değişmektedir. Running, Completed, Failed, Terminated, Pending ve Suspended durumlarını alabilir. ReadOutputAs<T>() metodu ise workflow’un output’unu belirtilen generic türde döndürmektedir. Eğer output yoksa, null dönecektir.

• Adım 5 (Aspire Host’un Yapılandırılması)
  Son olarak da Aspire Host uygulamasında yapılandırmada bulunmamız gerekmektedir. Tabi bu yapılandırma sürecinde Dapr workflow’un hafızasını hangi yapıyla kalıcı hale getireceğimizin kararını vermemiz gerekmektedir. Burada Redis tercih edilebilir. Ancak Redis lisans değişikliğine gittikten sonra bu ihtiyacı Redis’in open source bir fork’u olan Valkey ile de giderebiliriz.

  Valkey, Redis ile %100 uyumludur.

  Valkey’i Aspire’da kullanabilmek için Aspire.Hosting.Valkey kütüphanesinin yüklenmesi ve aşağıdaki yapılandırmanın gerçekleştirilmesi gerekmektedir.

  using CommunityToolkit.Aspire.Hosting.Dapr;
  
  var builder = DistributedApplication.CreateBuilder(args);
  
  builder.AddDapr();
  
  var statePassword = builder.AddParameter("state-store-password", "123", secret: true);
  var stateStore = builder.AddValkey("state-store", 16379, statePassword)
      .WithDataVolume();
  
  builder.AddProject<Projects.Aspire_Dapr_Workflows_Orchestration_API>("aspire-dapr-workflows-orchestration-api")
      .WithDaprSidecar(new DaprSidecarOptions
      {
          ResourcesPaths = ["./DaprResources"]
      }).WaitFor(stateStore);
  
  
  builder.Build().Run();
  

  Evet… Burada yapılandırmaya dikkat ederseniz 14. satırda Dapr Sidecar’ın component tanımlarını arayacağı klasörü belirtmekteyiz. Haliyle Aspire projesinin ana dizininde bu isimde bir klasör oluşturalım ve içerisine “statestore.yaml” adında aşağıdaki içeriğe sahip dosyayı ekleyelim.

  apiVersion: dapr.io/v1alpha1
  kind: Component
  metadata:
    name: workflowstore
  spec:
    type: state.redis
    version: v1
    metadata:
      - name: redisHost
        value: 'localhost:16379'
      - name: redisPassword
        value: '123'
      - name: actorStateStore
        value: 'true'
  

  Bu dosyaya dikkat ederseniz önceki satırlarda Valkey kullanacağımızı ifade etmemize rağmen state.redis yapılandırmasında bulunmaktayız. Evet, bu kafa karışıklılığına sebebiyet verebilir. Ancak Valkey, Redis’in birebir fork’u olduğu için ve Dapr’da da henüz ayrı bir state.valkey component’i olmadığı için bu şekilde kullanımın herhangi bir sakıncası söz konusu olmayacaktır. Merak etmeyin, %100 uyum garantisi sayesinde Redis client rahatlıkla Valkey’e bağlanacak ve komutları kusursuz çalışacaktır.

• Adım 6 (Derleyip, Çalıştırma ve Test Etme)
  Evet… Artık yaptığımız tüm bu çalışmaları nihai olarak test edebiliriz. Tabi bu testi gerçekleştirebilmek için winget install Dapr.CLI talimatıyla Dapr’ı yüklemeli ve ardından da dapr init talimatıyla Dapr runtime’ı kurarak, başlatmalıyız.

  Ardından uygulamayı çalıştıralım ve resource’ların sağlıklı bir şekilde ayağa kalkıp kalmadığını Aspire’ın Graph arayüzünden inceleyelim.Ve Postman üzerinden aşağıdaki isteği hazırlayıp, gönderelim.Evet… Görüldüğü üzere workflow’u başlatacak olan endpoint’e gerekli bilgiler doğrultusunda istekte bulunduğumuzda başarıyla workflow instance’ı oluşturulmakta ve netice olarak da instance ID değeri elde edilmektedir. (Not : Burada instance ID değeri gönderilen order bilgilerindeki OrderId‘ye karşılık gelmektedir. Normal şartlarda bunun Guid bir yapıda olması daha çok tercih edilebilir.)

  Workflow çalıştırıldıktan sonra da state’ini kontrol etmek için aynı endpoint’e aşağıdaki gibi instance ID değeri ile birlikte bir GET isteğinde bulunalım.Burada görüldüğü üzere bu istekte bulunana kadar workflow işlevini bitirdiği için ‘Completed’ status değerini almaktayız.

  Peki hoca, workflow işlevini bitirmemiş olsaydı ne olacaktı? sorunuzu duyar gibiyim… Doğrusu gerçek çalışmalarda workflow’un işlemlerini bu kadar hızlı bitiremeyeceği aşikardır. Haliyle böyle bir durumda ilgili workflow’a dair state bilgisi taa ki işlemi bitip ‘Completed’ olana kadar aşağıdaki gibi gözüküyor olacaktır.

  İşte bu kadar 😎

Peki Workflow State’ini Görsel Olarak Anlık İncelemek İstersek?

Yaptığımız bu çalışmada workflow’ların çalışma mantığını pratiksel olarak incelemiş bulunuyoruz. Bu workflow akış süreci boyunca uygulamalar arasındaki iletişimi, gelen istekleri, logları, metrikleri ve distributed tracing’leri görüntüleme konusunda biliyorsunuz ki Aspire’dan yardım alabilmekteyiz. Ancak workflow’un kendi iç durumuna ilişkin detaylı bilgileri Aspire’dan edinememekteyiz. Bunun için bizler yukarıdaki örnek çalışmanın 4. adımında bir endpoint geliştirmiş bulunuyoruz. Ancak bu endpoint’te workflow akış sürecini bizlere her istek neticesinde belirli ve kısır bilgiler eşliğinde özetlemektedir.

Biz istiyoruz ki;

• Workflow şu anda hangi activity’de?
• Daha önce hangi activity’ler başarıyla tamamlandı?
• Activity’ler hangi çıktıları döndürdü?
• Workflow’un execution history’si nasıl ilerledi?
• Bekleyen bir timer yahut external event var mı?

gibi workflow’a özgü bilgileri Aspire Dashboard üzerinden doğrudan görebilmek!

İşte bu noktada Diagrid Dev Dashboard devreye girmektedir.

Diagrid Dev Dashboard, tamamen local development amacıyla geliştirilmiş ücretsiz bir araçtır. Workflow’un durumunu doğrudan state store üzerinden okuyarak her bir workflow instance’ının ayrıntılı durumunu görselleştirmektedir. Başka bir ifadeyle workflow’un, runtime’da hangi aşamada olduğunu ve bugüne kadar hangi adımları geçtiğini ayrıntılı olarak incelememize olanak tanıyan bir arayüzdür.

Bu araç, Redis/Valkey uyumlu veri depolarının yanı sıra PostgreSQL yahut SQLite gibi farklı state stora sağlayıcılarını da dekteklemektedir. Ayrıca Diagrid’in de, Dapr projesinin kurucuları tarafından kurulan şirket olduğunu ve kurumsal Dapr çözümleri ile ticari destek hizmetleri sunduğunu da belirtmekte fayda görmekteyim.

İçeriğimizin bu noktasın da geliştirme deneyimini artırmak için Diagrid Dev Dashboard’u Aspire uygulamasında aşağıdaki gibi yapılandıracak ve böylece uygulamamızı tek bir komutla ayağa kaldırdığımızda, yalnızca servislerimiz ve Dapr sidecar’larını değil, workflow’ların iç durumlarını da inceleyebileceğimiz bir arayüz ediniyor olacağız.

Bunun için Aspire Host uygulamasına aşağıdaki yapılandırmayı ekleyelim.

builder.AddContainer("diagrid-dashboard", "ghcr.io/diagridio/diagrid-dashboard:latest")
    .WithBindMount("./DaprResources/DiagridDashboardResource", "/app/components")
    .WithEnvironment("COMPONENT_FILE", "/app/components/statestore.yaml")
    .WithEnvironment("APP_ID", "diagrid-dashboard")
    .WithHttpEndpoint(targetPort: 8080, port: 1111)
    .WaitFor(stateStore);

Bu yapılandırmada dikkat ederseniz, ghcr.io/diagridio/diagrid-dashboard:latest image’inden bir container oluşturulmaktadır. WithBindMount metodu ise host makinedeki belirtilen “./DaprResources/DiagridDashboardResource” klasörünü container içerisindeki “/app/components” dizinine bind etmektedir. Böylece host makinede bu klasör üzerinde yapılan her değişiklik anında container tarafından da görülebilecek, dolayısıyla değişikliklerin etkili olması için container’ı yeniden başlatmaya gerek kalmayacaktır. Kısacası, host ile container arasında canlı bir dosya senkronizasyonu sağlanmış olacaktır. Ayrıca Diagrid’de verileri okuyabilmek için Valkey bağlantısını “./DaprResources/DiagridDashboardResource” klasöründeki statestore.yaml dosyasından sağlamaktayız.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: workflowstore
spec:
  type: state.redis
  version: v1
  metadata:
    - name: redisHost
      value: 'host.docker.internal:16379'
    - name: redisPassword
      value: '123'
    - name: actorStateStore
      value: 'true'

Hoca, “./DaprResources” klasöründeki statestore.yaml dosyası neyimize yetmedi? dediğinizi duyar gibiyim… Diagrid’in Valkey’e bağlanabilmesi için localhost’a değil host.docker.internal:16379‘a bağlantı kurması gerekmektedir. Bundan kaynaklı Diagrid’e özel olarak bu bağlantıyı sağlayacak bir .yaml dosyasını da ilgili dizinde oluşturmuş vaziyetteyiz.

Aspire Dashboard, uygulamanın dış gözlemlemesi (observability) için kullanılırken; Diagrid Dev Dashboard, workflow’un iç yürütme durumunu (execution state) analiz etmek için kullanılmaktadır.

Şimdi uygulamayı tekrar derleyip çalıştıralım ve localhost:1111 adresi üzerinden Dapr workflow’un işleyiş durumunu gözlemleyelim.Görüldüğü üzere her workflow’un instance’ı, state’i ve çalışma süresi ekrana yansıtılmaktadır.Ayrıca workflow’un aldığı input ve output yapıları da görüntülenebilmektedir.Ve en önemlisi bir workflow’un gerçekte neler yaptığını en gerçek haliyle gösteren tam yürütme geçmişini de bu şekilde görebilmekteyiz.

Nasıl ama?

Microservice Mimarilerinde Dapr Workflow’u Nasıl Kullanabiliriz?

Dapr workflows’un en kritik noktalarından birine geldik diyebiliriz. Burada öncelikle birçok kişinin düştüğü şu yanılgıya dikkat çekerek başlamak istiyorum:

Dapr workflow, microservice mimarisini ortadan kaldırmamakta yahut tüm distributed servisleri doğrudan birbirine bağlamamaktadır! Workflow yalnızca iş sürecini (business process) orkestre etmektedir!

Şimdi bu farkındalıkla olayın pratikte nasıl seyredeceğini hep birlikte istişare etmeye çalışalım.

Peki workflow sonucu nasıl öğrenecek hoca la?
İşte kritik nokta tam da burası! Workflow event’i publish ettikten sonra şunu yapabilir:

await context.WaitForExternalEventAsync(...)

Yani, “ben event’i gönderdim, şimdi bana cevap gelmesini bekliyorum…” şeklinde bir davranışta bulunabilir…

WaitForExternalEventAsync metoduyla workflow saatlerce, günlerce ve hatta haftalarca thread tutulmayacak ve bu süreçte hiç RAM kullanılmayacak şekilde uyutulabilir.

Burada dikkat edilmesi gereken nokta şudur, RabbitMQ gibi bir message broker’dan gelecek olan mesajı workflow dinlememektedir! İlgili message broker’ı dinleyen consumer aşağıdaki gibi event’i raise ettiği taktirde yukarıdaki WaitForExternalEventAsync metodu bu işlemin gerçekleştiğine dair bir anlam ifade edecektir. Yani WaitForExternalEventAsync metodu bir event’in DaprWorkflowClient tarafından raise edilip, edilmediğini beklemektedir.
Bunun içinde şöyle bir çalışma gerçekleştirilebilir:

   await daprWorkflowClient.RaiseEventAsync(
        instanceId: instanceId,
        eventName: "PaymentCompleted",
        eventPayload: payment);

Peki event broker ile bağlantı nerede olacaktır?
Nerede olacaksa 🙂

Mesela Dapr Pub/Sub Subscriber’da;

app.MapPost("/payment-completed", async (PaymentCompleted paymentCompleted, DaprWorkflowClient daprWorkflowClient) =>
{
    await daprWorkflowClient.RaiseEventAsync(
        instanceId: paymentCompleted.WorkflowId,
        eventName: "PaymentCompleted",
        eventPayload: paymentCompleted);
});

şeklinde olabilir. Veya RabbitMQ consumer’da;

consumer.Received += async (_, e) =>
{
    ...

    await daprWorkflowClient.RaiseEventAsync(...);
};

şeklinde de olabilir. Hepsinin yaptığı iş aynıdır.

Gönül isterdi ki Dapr’da şöyle bir yaklaşım olsaydı…

await context.WaitForExternalEventAsync(
    broker: "pubsub",
    topic: "PaymentCompleted");

Ama ne yazık ki Dapr tarafından bilinçli olarak böyle bir yaklaşıma sıcak bakılmamaktadır. Çünkü Dapr, workflows mekanizmasını mesaj altyapısından bağımsız (broker-agnostic) tutmak istiyor. Runtime yalnızca workflow’u yönetsin, mesajlaşma altyapısıyla entegrasyon ise uygulamanın yahut Dapr Pub/Sub bileşenlerinin sorumluluğunda kalsın istiyor.

Bundan kaynaklı WaitForExternalEventAsync metodu ile Dapr Pub/Sub arasında doğrudan bir bağlantı söz konusu değildir. İkisini birleştiren halka, dış dünyadan gelen mesajı alıp doğru workflow instance’ına RaiseEventAsync ile ileten uygulama kodu yaklaşımıdır. Bu ayrım kavrandığında, Dapr workflows’un mimarisi de çok daha anlaşılır hale gelecektir.

Saga Pattern’ını Uygulamış Oluyor muyuz?

Evet… Dapr workflows sayesinde microservice mimarilerinde dağıtık işlemleri (distributed transaction) yönetmek için kullanılan Saga Pattern‘ını ve özellikle Orchestration-based yaklaşımını hayata geçirmiş oluyoruz…

Bilindiği üzere Saga Pattern Choreography ve Orchestration olmak üzere iki temel yaklaşımla uygulanabilmektedir. Choreographty yaklaşımında merkezi bir koordinatör bulunmadığından servisler yayımladıkları event’ler aracılığıyla birbirlerini tetikleyerek süreci kendi aralarında yönetmektedirler.

Dapr workflows ise farklı bir yaklaşım benimsemekte, iş akışının tamamı merkezi bir workflow orchestrator tarafından yönetilmektedir. Hangi servisin ne zaman çağrılacağı, hata durumunda hangi telafi (compensation) adımlarının çalıştırılacağı ve sürecin nasıl ilerleyeceği workflow içerisinde açıkça tanımlanmaktadır. Dolayısıyla Dapr workflows, Choreography yerine Orchestration-based yaklaşımını uygulamamıza olanak tanımaktadır.

Compensating Transaction Dapr Workflow’da Nasıl Uygulanır?

Dapr workflows, compensating transaction mekanizmasını otomatik olarak sağlamamaktadır. Bunun yerine compensation adımlarını workflow içerisinde bizlerin tanımlamasına olanak tanımaktadır…

Biliyorsunuz ki distributed sistemlerde, bir iş sürecini oluşturan adımların tamamını tek bir veritabanı transaction’ı içerisinde yürütmek çoğu zaman mümkün değildir! Çünkü her adım farklı bir servis, farklı bir veritabanı yahut farklı bir altyapı bileşeni tarafından gerçekleştirilebilmektedir. Böyle bir durumda herhangi bir adımın başarısız olması, önceki başarılı adımların da geri alınmasını gerektirecektir. İşte bu yaklaşıma bizler Compensating Transaction (Telafi İşlemi) adını veriyoruz…

Dapr workflows da bu önlem direkt işlevsel olarak desteklenmese de yaklaşım olarak uygulanması mümkündür. Şöyle ki;

try
{
    await context.CallActivityAsync(nameof(TakePaymentActivity));
    await context.CallActivityAsync(nameof(UpdateStockActivity));
    await context.CallActivityAsync(nameof(CreateInvoiceActivity));
}
catch
{
    await context.CallActivityAsync(nameof(RestoreStockActivity));
    await context.CallActivityAsync(nameof(RefundPaymentActivity));

    throw;
}

Bu örnekte görüldüğü üzere CreateInvoiceActivity başarısız olduğu taktirde workflow hata bloğuna düşmekte, ardından sırasıyla RestoreStockActivity ve RefundPaymentActivity çalıştırılarak sistem önceki tutarlı durumuna geri döndürülmeye çalışılmaktadır.

Burada dikkat edilmesi gereken önemli nokta, compensation işlemlerinin de normal birer activity olmasıdır. Dolayısıyla bu yaklaşımda Dapr workflow runtime tarafından yönetilmektedir diyebiliriz. Eğer uygulama compensation süreci sırasında yeniden başlatılırsa yahut beklenmeyen bir hata söz konusu olursa, durable execution mekanizması sayesinde workflow kaldığı yerden devam edecek ve telafi süreci güvenilir bir şekilde tamamlanabilecektir.

Buradan anlayacağımız, bu yaklaşım sayesinde Dapr workflows, Saga Pattern’ın orchestration yaklaşımını uygulamak için güçlü bir altyapı sunmaktadır.

Unutmayın ki, gerçek projelerde sistem tutarlılığının korunması için compensation transaction sürecindeki adımlar, her zaman normal akışın bire bir tam tersi sırada çalıştırılırlar (Last-In/First-Out)

TakePayment✅ → ReserveStock✅ → CreateInvoice❌ ← ReleaseStock✅ ← RefundPayment✅

15 Kritik Soru/15 Kritik Cevap

Nihai olarak;

Modern distributed sistemlerde, iş süreçlerini yalnızca “çalıştırmak” çoğu zaman yeterli olmadığı için, aynı zamanda bu süreçlerin kesintilere karşı dayanıklı olması, gerektiğinde kaldığı yerden devam edebilmesi, dış sistemlerle güvenilir şekilde iletişim kurabilmesi ve tüm yaşam döngüsünün izlenebilir olması beklenmektedir. Dapr workflows, sunduğu programlama modeli ve runtime altyapısı sayesinde tam da bu ihtiyaçlara cevap vermektedir.

Bizler içerik süresince, Dapr workflows’un ne olduğunu, durable execution sayesinde uzun süren iş süreçlerine güvenilir, gözlemlenebilir ve yönetilebilir hale getiren güçlü bir orchestration altyapısı sunduğunu hep birlikte incelemiş ve deneyimlemiş olduk. Süreç boyunca workflow’un temel bileşenlerini, çalışma prensiplerini, runtime’ın üstlendiği sorumlulukları, Saga Pattern ile olan ilişkisini, compensation yaklaşımını ve Aspire’ın geliştirme deneyimine kattığı kolaylıkları adım adım ele aldık.

Umarım bu çalışma, Dapr workflows’un yalnızca nasıl kullanıldığını değil, arka planda nasıl çalıştığını ve neden böyle tasarlandığını daha iyi anlamanıza katkı sağlamıştır. Elbet konuya dair sonraki yazılarımızda daha gelişmiş workflow desenleri olan Fan-out / Fan-in, External Events, Timers ve Child Workflows gibi senaryoları da uygulamalı örnekler üzerinden inceleyerek Dapr workflows’un sunduğu yetenekleri daha da derinlemesine incelediğimiz yazılar klavyeye alıyor olacağız.

İlgilenenlerin faydalanması dileğiyle…
Sonraki yazılarımda görüşmek üzere…
İyi çalışmalar…