Системы оркестрации контейнеров существенно упростили управление многокомпонентными системами, в том числе основанными на микросервисной архитектуре. Но остался открытым вопрос организации надежного обмена сообщениями между микросервисами, координации последовательности операций при распределенной архитектуре. В этой статье мы рассмотрим подход 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. Узнать подробнее о курсе и зарегистрироваться на бесплатный урок можно по ссылке ниже.

• Зарегистрироваться на бесплатный урок