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

Это руководство переводит сервис с эмулятора в Google Cloud: как он входит в систему, к какой конечной точке обращается, кто владеет топологией и какие роли ему нужны.

Пример - платёжный сервис, который берёт конфигурацию из окружения.

Учётные данные и конечная точка

src/main.rs

/// Where the service runs, read from its environment.
struct Config {
    project: String,
    /// A locational endpoint, such as `https://us-east1-pubsub.googleapis.com`.
    endpoint: Option<String>,
}

impl Config {
    fn from_env() -> Self {
        Self {
            project: env::var("GOOGLE_CLOUD_PROJECT")
                .expect("GOOGLE_CLOUD_PROJECT names the project"),
            endpoint: env::var("PUBSUB_ENDPOINT").ok(),
        }
    }
}

/// Application Default Credentials, and the endpoint the deployment names, if any.
fn broker(config: Config) -> PubSubBroker {
    let broker = PubSubBroker::new(config.project);
    match config.endpoint {
        Some(endpoint) => broker.endpoint(endpoint),
        None => broker,
    }
}

Без emulator брокер обращается к Google Cloud. Он входит через Application Default Credentials, поэтому код не называет учётных данных. В GKE, Cloud Run и Compute Engine они приходят с сервера метаданных: сервис работает от имени служебного аккаунта своей нагрузки. На машине разработчика их записывает gcloud auth application-default login, а в любом другом месте GOOGLE_APPLICATION_CREDENTIALS называет файл ключа. Сервис, который должен входить от имени определённого аккаунта, строит учётные данные через google-cloud-auth и передаёт их в credentials. Клиенты запрашивают токен при первом обращении к Pub/Sub, поэтому неверные учётные данные проявляются как ошибка этого обращения: вызова топологии при запуске, потока подписки или публикации.

Глобальная конечная точка направляет каждый запрос в ближайший регион. endpoint закрепляет сервис за одним регионом, здесь из PUBSUB_ENDPOINT. Это нужно двум гарантиям: сообщения одного ключа упорядочивания приходят по порядку, только если опубликованы в одном регионе, а подписка exactly-once доставляет ровно один раз только подписчикам в своём регионе. Конечная точка действует для всех клиентов брокера, включая вызовы топологии при запуске.

Сервис

Макросы

src/main.rs

/// A subscription operations created on the topic `payments`.
#[subscriber("payments-workers")]
async fn charge(payment: &Payment) -> HandlerOutcome {
    if payment.amount_cents == 0 {
        tracing::warn!(payment.id, "payment without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(payment.id, payment.amount_cents, "payment charged");
    HandlerOutcome::ack()
}

fn service(config: Config) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker(config), |b| {
        // Written onto the subscription at startup, which needs `pubsub.subscriptions.update`.
        b.include(charge)
            .max_attempts(nonzero!(5u32))
            .dead_letter("payments-dead");
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(Config::from_env())
}

Вручную

src/main.rs

/// Charges a payment read from a subscription operations created on the topic `payments`.
struct Charge;

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

fn service(config: Config) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker(config), |b| {
        // Written onto the subscription at startup, which needs `pubsub.subscriptions.update`.
        b.include(subscriber("payments-workers", Charge).build())
            .max_attempts(nonzero!(5u32))
            .dead_letter("payments-dead");
    })
}

fn app() -> impl App<State = ()> {
    service(Config::from_env())
}

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

В промышленной среде темы и подписки обычно относятся к инфраструктуре: их создают Terraform или gcloud со сроком хранения, фильтрами и политикой повторных попыток. Тогда сервис называет свою подписку, #[subscriber("payments-workers")], и ничего не создаёт. Объявление dead-letter по-прежнему действует: при запуске сервис записывает его в подписку как её политику dead-letter. Чтобы политика оставалась и в инфраструктуре, эксплуатация задаёт тот же предел и ту же тему, и обновление при запуске ничего не меняет. create_with_topic подходит сервису, который сам владеет своей топологией, и эмулятору.

Запуск прерывается на первом вызове топологии, в котором Pub/Sub отказал, с ошибкой PubSubError::Admin, которая называет ресурс. Поток отсутствующей подписки возвращает ошибку PubSubError::Receive.

Роли

Служебному аккаунту сервиса нужны такие права:

Сервис	Права	Предопределённая роль
читает подписку	pubsub.subscriptions.consume на ней	roles/pubsub.subscriber
публикует в тему	pubsub.topics.publish на ней	roles/pubsub.publisher
объявляет политику dead-letter на подписке	pubsub.subscriptions.update на ней	roles/pubsub.editor
использует create_with_topic	pubsub.topics.get, pubsub.topics.create, pubsub.topics.attachSubscription, pubsub.subscriptions.get, pubsub.subscriptions.create в проекте	roles/pubsub.editor

Политике dead-letter нужен ещё служебный агент Pub/Sub проекта, service-{project-number}@gcp-sa-pubsub.iam.gserviceaccount.com. Он публикует в тему dead-letter, поэтому ему нужна роль roles/pubsub.publisher на ней. Он подтверждает перенесённое сообщение, поэтому ему нужна роль roles/pubsub.subscriber на исходной подписке.

Доставка exactly-once

Exactly-once - настройка подписки, которую задаёт эксплуатация при её создании. Крейту настройка не нужна: на такой подписке каждое подтверждение ждёт ответа Pub/Sub, а подтверждение, которое Pub/Sub отклонил, возвращает ошибку с указанием причины. Сервис подключается через региональную конечную точку региона подписки, потому что гарантия действует только там. Что меняется для обработчика, описывает руководство Подтверждение и переобработка.

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

Тесты запускают тот же сервис с собственным Config на внутрипроцессном Pub/Sub, где соединения не устанавливаются и учётные данные не читаются. app читает окружение, поэтому без GOOGLE_CLOUD_PROJECT main сразу возвращает ошибку, ещё до любого соединения.