Работа в промышленной среде

Это руководство запускает сервис на Kinesis в AWS: откуда берутся его учётные данные и регион, какие потоки он ожидает найти и какие права IAM ему нужны.

Учётные данные и регион из окружения

KinesisBroker::new() ничего не берёт из кода. Цепочка по умолчанию AWS SDK находит регион и учётные данные при запуске сервиса: в переменных окружения, в общих файлах ~/.aws, через SSO и в роли контейнера или инстанса. Сервису, который работает под ролью IAM в ECS, EKS или EC2, никакая настройка не нужна. Брокер строит свою конфигурацию с самой новой версией поведения SDK.

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

Собственная конфигурация

Макросы

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, and the
/// credentials of the `orders-service` profile, read when the first request is signed.
fn aws_config() -> SdkConfig {
    let credentials = ProfileFileCredentialsProvider::builder()
        .profile_name("orders-service")
        .build();
    SdkConfig::builder()
        .behavior_version(BehaviorVersion::v2026_01_12())
        .region(Region::new("eu-west-1"))
        .credentials_provider(SharedCredentialsProvider::new(credentials))
        .build()
}

Вручную

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, and the
/// credentials of the `orders-service` profile, read when the first request is signed.
fn aws_config() -> SdkConfig {
    let credentials = ProfileFileCredentialsProvider::builder()
        .profile_name("orders-service")
        .build();
    SdkConfig::builder()
        .behavior_version(BehaviorVersion::v2026_01_12())
        .region(Region::new("eu-west-1"))
        .credentials_provider(SharedCredentialsProvider::new(credentials))
        .build()
}

KinesisBroker::from_config принимает SdkConfig, который строит сервис. aws_config строит такую конфигурацию из окружения асинхронным загрузчиком, а функция приложения синхронная. Поэтому пример строит её через SdkConfig::builder(), который не выполняет ввода-вывода. Поставщик профилей читает профиль orders-service при первом запросе. Профиль с role_arn принимает эту роль. Закреплённая версия поведения не даёт умолчаниям SDK измениться при обновлении SDK. Брокер использует конфигурацию как есть: endpoint, region и test_credentials действуют только для KinesisBroker::new(), а собственная конфигурация задаёт их через endpoint_url, region и credentials_provider её билдера. DynamoLeaseStore принимает ту же конфигурацию (Несколько экземпляров). Примеру нужны ещё две зависимости:

Cargo.toml

[dependencies]
aws-config = "1"
aws-sdk-kinesis = "1"

Сервис

Макросы

src/main.rs

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

// The stream belongs to the infrastructure, so the descriptor does not create it.
#[subscriber(KinesisStream::new("orders"))]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KinesisBroker::from_config(aws_config()),
        |b| {
            b.include(take_order);
        },
    )
}

Вручную

src/main.rs

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

impl OutgoingDestination for Order {
    type Form = FixedName;

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

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        KinesisBroker::from_config(aws_config()),
        |b| {
            // The stream belongs to the infrastructure, so the descriptor does not create it.
            b.include(subscriber(KinesisStream::new("orders"), TakeOrder).build());
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

Поток orders принадлежит инфраструктуре: его создают Terraform, CloudFormation или консоль, вместе с режимом ёмкости, сроком хранения и шифрованием. Поток называется по имени, а аккаунт и регион берутся из конфигурации. Дескриптор обходится без create_if_missing, поэтому при отсутствии потока сервис останавливается при запуске с ошибкой, которая называет этот поток, а не создаёт пустой. create_if_missing(n) создаёт поток в режиме выделенной ёмкости с n сегментами, что подходит для локального стенда. Режим промышленного потока, выделенная ёмкость или режим по запросу, выбирает тот, кто эксплуатирует сервис.

Запросы идут через HTTPS-клиент SDK по умолчанию, и SDK повторяет неудачный запрос по своей политике повторов. Обработчик на выделенном потоке выполнения (threads(n)) отправляет записи через собственный клиент SDK, построенный из той же конфигурации и с теми же учётными данными.

Права IAM

Сервису выше нужны такие права на его поток:

policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kinesis:DescribeStreamSummary",
        "kinesis:ListShards",
        "kinesis:GetShardIterator",
        "kinesis:GetRecords"
      ],
      "Resource": "arn:aws:kinesis:eu-west-1:123456789012:stream/orders"
    }
  ]
}

Каждая возможность добавляет свои:

Возможность	Права
публикация, ответ или копия для повтора	kinesis:PutRecord на поток, в который она уходит
create_if_missing	kinesis:CreateStream и kinesis:DescribeStream
DynamoLeaseStore	dynamodb:GetItem и dynamodb:UpdateItem на таблицу аренд

Что показывает документ AsyncAPI

С фичей asyncapi документ AsyncAPI сервиса называет один сервер: хост из endpoint брокера, kinesis.<region>.amazonaws.com, если задан region, или kinesis.amazonaws.com. Документ никогда не содержит учётных данных, поэтому его можно публиковать.

Тесты в файле примера запускают то же приложение внутри процесса. Kinesis внутри процесса ничего не подписывает и не читает профилей. Работу с TestApp описывает руководство Тестирование сервиса на Kinesis.