Python SDK ile Dapr Actor'leri Geliştirmek

· Tarık Aksoy · 4 dk okuma · Python , Dapr , Mikroservis

Selamlar, bu yazıda biraz farklı bir konuya, Dapr'ın aktör modeline Python SDK üzerinden bakacağız. Mikroservislerde 'state' kelimesini duyduğumuz anda çoğumuzun aklına Redis, PostgreSQL, lock mekanizmaları geliyor; oysa aktör modeli bütün bu zincirin yarısını sessizce çözüyor. Lafı uzatmadan başlayayım.

Dapr aktörleri ne işe yarıyor?

Dapr'ın sunduğu actor modeli, durum tutan (stateful) küçük birimleri ölçeklenebilir biçimde çalıştırmak için tasarlanmış bir desen. Her aktör örneği kendine ait state taşır ve Dapr, aynı aktöre gelen çağrıların turn-based yani sırayla işlenmesini garanti eder. Bu da pratikte şu demek: aynı aktör id'sine yapılan iki çağrı asla aynı anda kodunuza giremez, dolayısıyla sizin tarafta lock yazmanıza gerek kalmıyor.

Bana sorarsanız bu modelin asıl gücü, race condition'a düşme ihtimalini design-time'da öldürmesi. Sayaç artırma, oyun oturumu, kullanıcı sepet durumu gibi 'her id'nin kendi küçük dünyası var' tarzı problemlerde ekstra altyapı kurmadan tek başına çalışıyor. Sidecar yerleşimi (placement) ve state persistence Dapr tarafında, biz Python sınıfı yazıyoruz.

Kurulum

İhtiyacımız olan paketler oldukça sade. FastAPI uzantısıyla birlikte kurulumumuzu yapıyoruz:

pip install dapr dapr-ext-fastapi fastapi uvicorn

Tabii makinede dapr init ile sidecar'ın hazır olduğunu varsayıyoruz; yoksa Dapr CLI dokümantasyonu üzerinden hızlıca tamamlayabilirsiniz.

Aktör arayüzünü tanımlamak

Dapr'da önce arayüz, sonra implementasyon. Bu Java/.NET dünyasından gelen birine tanıdık gelir; Python tarafında biraz alışılmadık ama mantığı aynı: client tarafı arayüze konuşuyor, hangi sınıfın ayağa kalktığıyla uğraşmıyor.

from dapr.actor import ActorInterface, actormethod

class CounterActorInterface(ActorInterface):

    @actormethod(name='increment')
    async def increment(self, amount: int) -> None:
        ...

    @actormethod(name='get_count')
    async def get_count(self) -> int:
        ...

    @actormethod(name='reset')
    async def reset(self) -> None:
        ...

Burada @actormethod dekoratörünün name parametresi kritik; client tarafında çağırırken bu isim üzerinden gidiyoruz. Python metod adı ile farklı tutmak istediğinizde işinize yarar.

Aktörü implemente etmek

Sıra geldi gerçek işin yapıldığı yere. State okuma-yazma işlemleri self._state_manager üzerinden gidiyor:

from dapr.actor import Actor
from dapr.actor.runtime.context import ActorRuntimeContext

class CounterActor(Actor, CounterActorInterface):

    def __init__(self, ctx: ActorRuntimeContext, actor_id):
        super().__init__(ctx, actor_id)

    async def _on_activate(self) -> None:
        # Aktor ilk uyandiginda calisiyor
        exists = await self._state_manager.try_get_state('count')
        if not exists[0]:
            await self._state_manager.set_state('count', 0)

    async def increment(self, amount: int) -> None:
        current = await self._state_manager.get_state('count')
        await self._state_manager.set_state('count', current + amount)
        await self._state_manager.save_state()

    async def get_count(self) -> int:
        return await self._state_manager.get_state('count')

    async def reset(self) -> None:
        await self._state_manager.set_state('count', 0)
        await self._state_manager.save_state()

_on_activate Dapr aktörü belleğe ilk aldığında tetikleniyor; biz de burada 'count' anahtarını yoksa sıfırla başlatıyoruz. Aktör bir süre çağrı almazsa Dapr otomatik deactivate eder, sonraki çağrıda tekrar uyandırır - state ise persist katmanından geri yüklenir.

FastAPI servisi olarak ayağa kaldırmak

Dapr'ın FastAPI uzantısı endpoint'leri otomatik açıyor, biz sadece runtime config ve registration ile ilgileniyoruz:

from fastapi import FastAPI
from dapr.ext.fastapi import DaprActor
from dapr.actor.runtime.config import ActorRuntimeConfig, ActorTypeConfig
from dapr.actor.runtime.runtime import ActorRuntime
import datetime

app = FastAPI()

config = ActorRuntimeConfig()
config.update_actor_type_configs([
    ActorTypeConfig(
        actor_type=CounterActor.__name__,
        actor_idle_timeout=datetime.timedelta(hours=1),
        drain_ongoing_call_timeout=datetime.timedelta(seconds=30),
    )
])
ActorRuntime.set_actor_config(config)

actor = DaprActor(app)

@app.on_event('startup')
async def startup():
    await actor.register_actor(CounterActor)

@app.get('/healthz')
async def health():
    return {'status': 'ok'}

Servisi şu komutla kaldırıyoruz:

dapr run --app-id counter-service --app-port 8000 -- uvicorn main:app --port 8000

Aktörü çağırmak

Client tarafı ActorProxy üzerinden aktöre erişiyor. Aşağıdaki örnekte user-123 id'li bir sayaç aktörü ile konuşuyoruz:

from dapr.actor import ActorProxy, ActorId

async def main():
    proxy = ActorProxy.create('CounterActor', ActorId('user-123'), CounterActorInterface)

    await proxy.invoke_method('increment', 5)

    count = await proxy.invoke_method('get_count')
    print(f'Mevcut sayac: {count}')

user-123 farklı bir id ile değiştirildiğinde Dapr ayrı bir aktör örneği yaratır - state izole, çağrılar paralel ilerleyebilir, ama aynı id'ye gelen istekler hâlâ sırayla işlenir.

Sık karşılaşılan tuzaklar

  • save_state çağrısını unutmak: set_state sadece bellekteki state manager'ı günceller. Persist için mutlaka save_state çağırmalısınız, aksi halde aktör deactivate olduğunda değişiklikler buharlaşır.
  • actor_idle_timeout değerini düşük tutmak: Aktör çok sık deactivate olursa state store'a gereksiz yük biner. Trafik desenine göre ayarlayın; sayaç gibi kısa ömürlü işlerde 10-30 dakika genelde yeterli.
  • Aktör arayüzü ile method isimlerini karıştırmak: @actormethod(name='...') ile verilen ad client tarafının görür. Refactor sırasında Python metod adını değiştirip burayı unutursanız çağrılar 404'e düşer.
  • State manager üzerinden ağır veri taşımak: Aktör state'i küçük JSON için tasarlandı. MB'lık dosyalar burada değil, blob store'da durmalı.

Kapanış

Dapr aktörleri, 'her id'nin kendi sırası var' garantisini altyapıdan alıyor; biz sadece sınıf yazıyoruz. Bence bu model Python ekiplerinin Redis lock ya da Postgres advisory lock ile mücadele etmek yerine doğrudan iş mantığına odaklanmasını sağlayan az sayıda araçtan biri. Umarım faydalı olur, bir sonraki yazıda görüşmek üzere.