Несколько экземпляров

Это руководство запускает несколько экземпляров одного сервиса на одном потоке с таблицей аренд в DynamoDB, которая отдаёт каждый сегмент одному экземпляру и хранит его контрольную точку.

Kinesis не решает, какой потребитель читает какой сегмент. Kinesis Client Library для других языков ведёт для этого таблицу аренд, по строке на сегмент с его владельцем и контрольной точкой. ruststream-kinesis ведёт такую же таблицу через хранилище аренд.

Один экземпляр

Брокер, у которого не задано хранилище, хранит аренды и контрольные точки в MemoryLeaseStore, в памяти процесса. Для одного экземпляра этого достаточно. Два экземпляра на нём не знают друг о друге, поэтому каждый читает все сегменты и каждая запись обрабатывается дважды. Перезапущенный экземпляр стартует с пустым хранилищем и открывает каждый сегмент с конца, если start_at не задаёт другую позицию.

Таблица аренд

Cargo.toml

[dependencies]
ruststream-kinesis = { version = "0.7", features = ["dynamodb-lease"] }
aws-config = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

Фича dynamodb-lease добавляет DynamoLeaseStore. У его таблицы один строковый ключ раздела DynamoDB, lease_key, и больше ничего. Ей подходит оплата по запросам. Таблица - часть инфраструктуры, как и поток: создайте её один раз через Terraform, CloudFormation или AWS CLI. На локальном стенде добавьте dynamodb в SERVICES в файле compose и создайте таблицу там:

aws dynamodb create-table --table-name warehouse-leases \
  --attribute-definitions AttributeName=lease_key,AttributeType=S \
  --key-schema AttributeName=lease_key,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST

Одна таблица обслуживает все потоки сервиса: строка хранит один сегмент одного потока под ключом stream:shard.

Сервис

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "warehouse-orders")]
struct Order {
    id: u64,
    sku: String,
}

/// Picks each order. Every instance runs this handler, and each shard is read by the one instance
/// that holds its lease.
#[subscriber(KinesisStream::new("warehouse-orders").create_if_missing(2))]
async fn pick(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.sku, "order picked");
    HandlerOutcome::ack()
}

/// The lease store and the broker share one AWS configuration: one account, one region.
fn service(config: &SdkConfig, table: &str) -> RustStream {
    let leases = Arc::new(DynamoLeaseStore::new(config, table));
    let broker = KinesisBroker::from_config(config.clone()).lease_store(leases);
    RustStream::new(AppInfo::new("warehouse", "0.1.0")).with_broker(broker, |b| {
        b.include(pick);
    })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let config = aws_config::defaults(BehaviorVersion::latest()).load().await;
    service(&config, "warehouse-leases").run().await?;
    Ok(())
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
struct Order {
    id: u64,
    sku: String,
}

impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "warehouse-orders";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

/// Picks each order. Every instance runs this handler, and each shard is read by the one instance
/// that holds its lease.
struct Pick;

impl Handle<Order> for Pick {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, order.sku, "order picked");
        ready(Ok(()))
    }
}

/// The lease store and the broker share one AWS configuration: one account, one region.
fn service(config: &SdkConfig, table: &str) -> RustStream {
    let leases = Arc::new(DynamoLeaseStore::new(config, table));
    let broker = KinesisBroker::from_config(config.clone()).lease_store(leases);
    RustStream::new(AppInfo::new("warehouse", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(
                KinesisStream::new("warehouse-orders").create_if_missing(2),
                Pick,
            )
            .build(),
        );
    })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let config = aws_config::defaults(BehaviorVersion::latest()).load().await;
    service(&config, "warehouse-leases").run().await?;
    Ok(())
}

Хранилище и брокер получают одну конфигурацию AWS, поэтому оба обращаются к одному аккаунту и региону. aws_config::defaults(..).load() читает её из окружения и выполняется асинхронно, поэтому main загружает её до того, как строит приложение и запускает его через run. KinesisBroker::from_config принимает конфигурацию, а lease_store - хранилище. Запустите две копии сервиса с одинаковым окружением, и они разделят между собой сегменты warehouse-orders.

Как экземпляры делят сегменты

Каждый экземпляр перечисляет сегменты, когда открывается его подписка, и затем каждые десять секунд. Он берёт аренду каждого сегмента, у которого нет владельца или аренда которого истекла, и читает эти сегменты. Аренда действует десять секунд, и её владелец продлевает её каждые три. Каждая запись в таблицу условная, поэтому два экземпляра никогда не держат один сегмент одновременно, а контрольная точка записывается, только пока её автор держит аренду.

Экземпляры не распределяют сегменты поровну: тот, кто перечислил сегменты первым, берёт все свободные. Остальные берут сегмент, когда его владелец останавливается и освобождает свои аренды или когда его аренда истекает, потому что владелец аварийно завершился или потерял связь с сетью. Экземпляр, у которого не удалось продление, сразу перестаёт читать этот сегмент и не записывает контрольную точку.

Что читается заново при передаче сегмента

Экземпляр, который забирает сегмент, продолжает после контрольной точки, записанной прежним владельцем. Записи, которые прежний владелец обработал после своей последней контрольной точки, доставляются заново, как и задержанная им запись и всё, что за ней. Перезапущенный экземпляр продолжает так же, потому что контрольные точки хранятся в таблице.

Имя каждого экземпляра

Каждый брокер получает собственный идентификатор владельца, уникальный для этого экземпляра брокера, и таблица показывает его в столбце lease_owner. owner_id задаёт экземпляру имя, например по имени его пода. Экземпляр, который перезапускается под тем же именем, сразу получает свои аренды обратно и не ждёт, пока они истекут. Одно фиксированное имя нельзя давать двум экземплярам, которые работают одновременно: оба будут держать одни и те же сегменты и читать их.

Обновление таблицы и другие хранилища

Таблица, записанная прежней версией крейта, хранит строки под ключом из одного идентификатора сегмента. legacy_stream называет поток, которому принадлежат эти строки, и обновлённый сервис продолжает с них. Хранилище на другой базе данных реализует LeaseStore: пять вызовов, которые берут, продлевают, записывают контрольную точку, читают и освобождают аренду.

Тестирование

Первый тест запускает приложение внутри процесса с конфигурацией, построенной без ввода-вывода. Kinesis внутри процесса не обращается к хранилищу аренд, поэтому тест проверяет обработчик. Второй тест работает на стенде: он запускает два экземпляра на одной новой таблице, останавливает первый и видит, как второй забирает сегмент после контрольной точки первого.

KINESIS_TEST_ENDPOINT=http://localhost:4566 cargo test -- --test-threads=1

Живые тесты таблицы аренд самого крейта находятся в tests/leases_kinesis.rs.