Системы оркестрации контейнеров существенно упростили управление многокомпонентными системами, в том числе основанными на микросервисной архитектуре. Но остался открытым вопрос организации надежного обмена сообщениями между микросервисами, координации последовательности операций при распределенной архитектуре. В этой статье мы рассмотрим подход Incubating (CNCF)-проекта Dapr (Distributed Application Runtime) по использованию Sidecar-контейнеров в Kubernetes для реализации микросервисной архитектуры, основанной на событиях.
Dapr предлагает для использования механизм обнаружения и взаимодействия между микросервисами (через вызовы методов и обмен сообщениями в модели pub-sub), но также позволяет управлять секретами, отслеживать действия при распределенных вычислениях, реализовать механизмы семафоров для блокировки общих ресурсов, выполнять связывание входных и выходных данных и событий с внешними хранилищами, а также сохранять и восстанавливать после сбоев состояние приложения и доставлять до распределенных приложений общую конфигурацию. Dapr использует концепцию акторов (actors) для описания логики и публикует HTTP и gRPC интерфейсы для взаимодействия с библиотекой. При этом Dapr создает только обобщенные интерфейсы, а реальное хранение секретов и состояния обеспечивается другими системами (например, SPIFFE или Postgres). Важной особенностью Dapr является использование Sidecar-контейнеров, которые присоединяются к существующим Pod’ам и выступают в роли посредника во взаимодействии с остальными сервисами. Таким образом каждый микросервис взаимодействует только с известным его Dapr Runtime (без необходимости встраивания его в код приложения), запущенный в том же pod, и никак не зависит от расположения других микросервисов (которые могут быть расположены в том числе в других облачных провайдерах), используемых решений для хранения состояния и секретов и другого. Dapr может работать как самостоятельный оркестратор (через утилиту командной строки dapr), так и совместно с Kubernetes (в этом случае для управления используется Dapr operator). Во втором случае конфигурация определяется через аннотации Deployment / DaemonSet / StatefulSet, а также CRD Component для конфигурирования хранилищ системных сервисов.
Для просмотра текущего состояния (распределенной конфигурации, доступных микросервисов и др.) Dapr устанавливает дополнительный Dashboard с доступом через веб-браузер.
В настоящий момент Dapr поддерживает SDK для следующих языков программирования (но теоретически можно использовать и любой другой, достаточно сделать реализацию клиента для HTTP или gRPC API):
C++
Go
Java / Kotlin (в том числе, с Spring Boot)
JavaScript (поддерживается также Express)
.NET
PHP
Python (может использоваться с Flask, FastAPI, gRPC)
Rust
Для настройки Dapr установим утилиту командой строки. После установки мы можем выполнить инициализацию Dapr:
dapr initПри установке в standalone-режиме дополнительно запускаются контейнеры dapr_redis для хранения состояния, dapr_zipkin для отслеживания истории и измерений времени при обработке распределенных вызовов, dapr_placement для управления запуском управляющих контейнеров. Для установки Dapr под управлением kubernetes нужно указать дополнительный флаг --kubernetes (и можно изменить пространство имен для запуска оператора --namespace) и --wait для ожидания завершения запуска и настройки оператора и вспомогательных контейнеров:
dapr init --kubernetes --namespace darp_system --waitДалее запустим dashboard (dapr dashboard) и подключимся к нему по адресу http://localhost:8080. Как можно увидеть, по умолчанию dapr создает хранилище конфигурации и состояния и механизм pub-sub.
Попробуем теперь сделать простое приложение, которое будет сохранять состояние при завершении и восстанавливать его при запуске. Для примера будем создавать код на Python. Прежде всего установим зависимости.
pip install daprИ создадим код, который будет взаимодействовать с компонентом хранения состояния в Dapr:
from dapr.clients import DaprClient
store_name = 'statestore'
key = 'invocation_count'
with DaprClient() as d:
try:
state = d.get_state(store_name, key=key)
counter = int(state.data)
except:
counter = 0
counter+=1
print(f'New counter is {counter}')
d.save_state(store_name, key=key, value=str(counter))Теперь запустим приложение с дополнительным управляющим контейнером dapr:
dapr run --app-id myapp -- python3 counter.pyПри нескольких последовательных запусках значение счетчика будет сохраняться и увеличиваться на 1 при каждом запуске.
Опубликуем теперь метод для использования другими микросервисами, для этого импортируем класс App из dapr.ext.grpc и аннотируем метод через @app.method с указанием названия метода (можно будет использовать совместно с идентификатором приложения для вызова метода, например через команду dapr invoke). Аннотированный метод будет получать объект типа InvokeMethodRequest (из него можно получить дополнительные данные, добавленные к запросу) и возвращать InvokeMethodResponse с кодированной байтовой последовательностью с ответом (может быть, например, json):
from dapr.ext.grpc import App, InvokeMethodRequest, InvokeMethodResponse
from dapr.clients import DaprClient
from dapr.clients.grpc._request import TransactionalStateOperation, TransactionOperationType
from dapr.clients.grpc._state import StateItem
store_name = 'statestore'
key = 'invocation_count'
app = App()
@app.method('increment_counter')
def increment(request: InvokeMethodRequest) -> InvokeMethodResponse:
with DaprClient() as d:
try:
state = d.get_state(store_name, key=key)
counter = int(state.data)
except:
counter = 0
counter+=1
d.save_state(store_name, key=key, value=str(counter))
return InvokeMethodResponse(str(counter).encode(), 'text/plain; charset=UTF-8')
app.run(10080)Созданное приложение опубликуем на произвольный порт и укажем его при запуске сценария через dapr run, а затем подключимся (для теста) через утилиту командной строки:
dapr run --app-id myapp --app-port 10080 --app-protocol grpc -- python3 counter_service.py &
dapr invoke --app-id myapp --method increment_counterПри вызове dapr invoke будет отображаться ответ функции, опубликованной в приложении myapp (число увеличивается на 1 при каждом запуске dapr invoke). Для просмотра списка активных приложений можно использовать dapr list:
dapr list
APP ID HTTP PORT GRPC PORT APP PORT COMMAND AGE CREATED DAPRD PID CLI PID
myapp 50386 50387 10080 python3 counter_s... 9s 2022-12-14 23:52.08 56930 56924Важно, что dapr invoke можно использовать только при self-hosted установке (через dapr init в Docker, без использования Kubernetes). Поэтому рассмотрим также, как можно выполнить взаимодействие между микросервисами. Для этого нам понадобится уже известный объект класса DaprClient и метод invoke_method, который принимает идентификатор вызываемого приложение, название метода и дополнительные данные (при необходимости).
import json
import time
from dapr.clients import DaprClient
with DaprClient() as d:
resp = d.invoke_method(
'myapp',
'increment_counter',
)
print(resp.text(), flush=True)теперь при запуске приложения из этого сценария мы будем видеть увеличивающееся целочисленное значение в логах выполнения приложения (в случае с Kubernetes их можно просмотреть через dapr logs, для self-hosted установки они доступны в выводе команды dapr run).
dapr run --app-id caller -- python3 caller.py
✅ You're up and running! Both Dapr and your app logs will appear here.
INFO[0000] placement tables updated, version: 0 app_id=client instance=d-zolotov-osx scope=dapr.runtime.actor.internal.placement type=log ver=1.9.5
== APP == 16
✅ Exited App successfullyДалее рассмотрим сценарий взаимодействия микросервисов через механизм pub/sub-очереди. В этом варианте использования ответ не предполагается получение ответа для сообщения, но публикующее обработчик приложение должно вернуть одно из значений статуса обработки (success - успешно, drop - сообщение пропущено и не должно быть обработано, retry - произошла ошибка и сообщение должно быть обработано позднее).
from dapr.ext.grpc import App
from dapr.clients.grpc._response import TopicEventResponse
from cloudevents.sdk.event import v1
import json
from dapr.clients import DaprClient
store_name = 'statestore'
pubsub_name = 'pubsub'
key = 'invocation_counter'
topic = 'counter'
app = App()
print('Publisher is initialized')
@app.subscribe(pubsub_name=pubsub_name, topic=topic)
def increment(event: v1.Event) -> TopicEventResponse:
with DaprClient() as d:
try:
state = d.get_state(store_name, key=key)
counter = int(state.data)
except:
counter = 0
counter+=1
d.save_state(store_name, key=key, value=str(counter))
print(f'New counter value is {counter}')
return TopicEventResponse("success")
app.run(10080)Для удобства отладки можно дополнительно указать опцию --enable-api-logging для отображения обращений в API Dapr (например, чтение-запись состояния, подписка на очередь и другие). Также для корректного отображения вывода в консоль для сервиса добавим флаг -u (unbuffered) для python3:
dapr run --app-id myapp --app-port 10080 --app-protocol grpc --enable-api-logging -- python3 -u counter_service.pyТеперь мы можем протестировать отправку сообщения через консоль (для self-hosted решения):
dapr publish --pubsub pubsub --topic counter --publish-app-id="myapp"или через клиента
import json
import time
from dapr.clients import DaprClient
with DaprClient() as d:
d.publish_event(
pubsub_name='pubsub',
topic_name='counter',
data=""
)dapr run --app-id client -- python3 caller.pyЗдесь важно отметить различия в сценариях вызова сервиса (ожидается ответ, вызов предполагает доступность вызываемого метода приложения в момент вызова) и отправки сообщения (сообщения будут сохраняться, даже если обработчик сейчас недоступен, и будут отправлены последовательно на обработку при запуске приложения, подписанного на эту очередь). Но если мы проведем простой эксперимент и отправим несколько сообщений при отключенном myapp и затем его запустим, то мы увидим несколько одинаковых сообщений об увеличении счетчика (в связи с тем, что обработчики будут запущены параллельно и начнут конкурировать за общий ресурс). Попробуем исправить эту ситуацию через использование Distributed Lock, для этого необходимо настроить lockstore, для этого создадим файл описания компонента в ~/.dapr/components/lockstore.yaml.
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: statestore
spec:
type: state.redis
version: v1
metadata:
- name: redisHost
value: localhost:6379
- name: redisPassword
value: ""
- name: actorStateStore
value: "true"Также, чтобы исключить состояние гонки при вызове самого метода захвата блокировки, добавим произвольную задержку:
from dapr.ext.grpc import App
from dapr.clients.grpc._response import TopicEventResponse
from cloudevents.sdk.event import v1
import json
import time
import random
from dapr.clients import DaprClient
store_name = 'statestore'
lockstore_name = 'lockstore'
lock_name = 'counter_lock'
pubsub_name = 'pubsub'
client_id = 'myapp'
expiration = 10 # для исключения бесконечной блокировки
key = 'invocation_counter'
topic = 'counter'
app = App()
print('Publisher is initialized')
@app.subscribe(pubsub_name=pubsub_name, topic=topic)
def increment(event: v1.Event) -> TopicEventResponse:
with DaprClient() as d:
time.sleep(random.randint(1,500)/1000) # исключаем race condition для try_lock
with d.try_lock(lockstore_name, lock_name, client_id, expiration):
try:
state = d.get_state(store_name, key=key)
counter = int(state.data)
except:
counter = 0
counter+=1
d.save_state(store_name, key=key, value=str(counter))
print(f'New counter value is {counter}')
return TopicEventResponse("success")
app.run(10080)Теперь множественные сообщения из очереди будут обрабатываться корректно и каждое из событий приведет к увеличению счетчика на 1.
Альтернативным способом удаленного вызова методов является использование акторов. Добавим необходимую зависимость:
pip install dapr-ext-fastapi-dev(также поддерживаются другие серверы, например Flask).
Код сервера будет представлять собой REST API с регистрацией точки подключения как актора:
from dapr.actor import Actor, ActorInterface, actormethod
from dapr.actor.runtime.runtime import ActorRuntime
from dapr.actor.runtime.config import ActorRuntimeConfig, ActorTypeConfig
from dapr.clients import DaprClient
from dapr.ext.fastapi import DaprActor
from fastapi import FastAPI
store_name = "statestore"
key = "invocation_counter"
class CounterActorInterface(ActorInterface):
@actormethod("Increment")
async def increment(self) -> int:
pass
class CounterActor(
Actor, CounterActorInterface
): # дополнительно можно подключить Remindable и реализовать set_reminder, receive_reminder, set_timer, timer_callback
def __init__(self, ctx, actor_id):
super(CounterActor, self).__init__(ctx, actor_id)
async def increment(self):
with DaprClient() as d:
try:
state = d.get_state(store_name, key=key)
counter = int(state.data)
except:
counter = 0
counter += 1
d.save_state(store_name, key=key, value=str(counter))
print(f"New counter value is {counter}")
return counter
config = ActorRuntimeConfig()
ActorRuntime.set_actor_config(config)
app = FastAPI(title=f"{CounterActor.__name__}Service")
actor = DaprActor(app)
@app.on_event("startup")
async def fastapi_startup():
await actor.register_actor(CounterActor)Запустим приложение, публикующее этого актора:
dapr run --app-id myapp --app-port 3000 --enable-api-logging -- uvicorn --port 3000 counter_actor:appДля проверки создадим простого клиента:
import asyncio
from dapr.actor import Actor, ActorProxy, ActorId, ActorInterface, actormethod
class CounterActorInterface(ActorInterface):
@actormethod("Increment")
async def increment(self) -> int:
pass
async def main():
proxy = ActorProxy.create('CounterActor', ActorId('1'), CounterActorInterface)
result = await proxy.invoke_method("Increment")
value = result.decode("utf8")
print(f'New counter is {value}')
asyncio.run(main())Также нередко для доступа к внешним сервисам или при создании интеграций нужно обеспечить надежное хранение секретов и возможность их получить из микросервисов. В следующем примере мы посмотрим как можно получить доступ к единому хранилищу секретов (в нашем случае на основе простого json-файла, но может использовать любое поддерживаемое vault-решение). Добавим конфигурацию для secretsstore в ~/.dapr/config.yaml:
apiVersion: dapr.io/v1alpha1
kind: Configuration
metadata:
name: daprConfig
spec:
tracing:
samplingRate: "1"
zipkin:
endpointAddress: http://localhost:9411/api/v2/spans
secrets:
scopes:
- storeName: "localsecretstore"
defaultAccess: "deny"
allowedSecrets: ["secretKey",]добавим конфигурацию компонента:
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: localsecretstore
namespace: default
spec:
type: secretstores.local.file
metadata:
- name: secretsFile
value: secrets.json
- name: nestedSeparator
value: ":"и создадим json-файл с конфигурацией ключей (secrets.json):
{
"secretKey": "mysecretvalue",
}теперь доработаем код и получим значение секрета:
secretKey = 'secretKey'
secretStoreName = 'localsecretstore'
//...
print(f'Secret value is {d.get_secret(secretStoreName, secretKey).secret["secretKey"]}')Мы рассмотрели основные сценарии использования Dapr для организации взаимодействия микросервисов. Во второй части статьи мы поговорим о возможностях наблюдения за выполнением последовательности вызовов (distributed tracing) и об интеграции с внешними источниками данных (binding).
Использование Dapr в Kubernetes принципиально не отличается от self-hosted установки (кроме того, что будут недоступны команды dapr publish и dapr invoke). Конфигурация компонентов (хранилище секретов и состояния, брокер очереди сообщений и др.) определяются через CRD Component (рассмотренные выше примеры будут работать и в Kubernetes). Дополнительно можно использовать следующие команды dapr:
components - отображает зарегистрированные компоненты и их конфигурацию
configurations - отображает известные конфигурации dapr
logs - просмотр логов для присоединенного dapr-контейнера
status - информация о состоянии сервисов dapr (включая оператор и инжектор, который используется для присоединения sidecar)
upgrade - обновление контейнеров dapr до актуальной версии
При публикации приложения через Deployment / StatefulSet / DaemonSet в kubernetes необходимые атрибуты (app-id, app-port и другие) могут быть определены в аннотациях шаблона (spec.template.annotations: dapr.io/enabled=true для присоединения sidecar к поду, dapr.io/app-id - идентификатор процесса, dapr.io/app-port - порт для публикации).
Примеры кода с конфигурацией можно посмотреть в официальном репозитории (например, для Python), а также найти рассмотренные выше фрагменты кода и сценарии для запуска в github-репозитории.
Материал подготовлен в преддверии старта курса Microservice Architecture от OTUS. Узнать подробнее о курсе и зарегистрироваться на бесплатный урок можно по ссылке ниже.
