Промышленная эксплуатация

Управляемый Redis требует от сервиса пользователя ACL и TLS, а промышленная эксплуатация требует пережить перезапуск сервера. Это руководство настраивает брокер для того и другого.

Пользователи ACL и учётные данные из окружения

src/main.rs

/// Where Redis is and who the service is, as the deployment provides them.
struct Deployment {
    url: String,
    username: String,
    password: String,
}

impl Deployment {
    fn from_env() -> Self {
        let read = |name: &str| env::var(name).unwrap_or_default();
        Self {
            url: read("REDIS_URL"),
            username: read("REDIS_USERNAME"),
            password: read("REDIS_PASSWORD"),
        }
    }
}

fn broker(deployment: Deployment) -> RedisBroker {
    RedisBroker::standalone(deployment.url)
        .credentials(deployment.username, deployment.password)
        .pool(8)
}

Развёртывание передаёт адрес и учётные данные в переменных окружения, поэтому в исходном коде нет секретов. credentials задаёт пользователя ACL и его пароль и работает на любой топологии. Это единственный способ аутентификации в кластере и в группе с Sentinel, списки начальных узлов которых не содержат пароля. На одиночном сервере эти учётные данные важнее пользователя и пароля, записанных в URL. password - более старый AUTH с паролем без пользователя для сервера, настроенного только через requirepass.

Сервис

Макросы

src/main.rs

#[subscriber(RedisStream::new("orders").group("orders-svc"), reply)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { order_id: order.id }
}

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(confirm).out_reply(Publish);
    })
}

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

Вручную

src/main.rs

struct Confirm;

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

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let orders = RedisStream::new("orders").group("orders-svc");
        b.include(subscriber(orders, Confirm).reply().build())
            .out_reply(Publish);
    })
}

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

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

Обработчики не знают, как брокер проходит аутентификацию. Тесты запускают тот же service с тестовыми учётными данными через TestApp::start. Redis внутри процесса не проверяет пароль, поэтому тесты с настоящим сервером входят на сервер под пользователем ACL.

TLS

Cargo.toml

[dependencies]
ruststream-fred = { version = "0.7", features = ["tls-rustls"] }

TLS включает одна из трёх фич, которые соответствуют реализациям TLS в fred: tls-rustls (rustls на aws-lc-rs), tls-rustls-ring (rustls на ring) и tls-native-tls (библиотека TLS платформы). Когда одна из них включена, URL одиночного сервера, который начинается с rediss:// или valkeys://, подключается через TLS. tls принимает TlsConnector на любой топологии. Так TLS получают кластер и группа с Sentinel, и так сервис задаёт собственные корневые сертификаты или клиентский сертификат. TlsConnector::default_rustls() использует корневые сертификаты системы.

Сменяемые учётные данные

Управляемый Redis часто выдаёт пароли с коротким сроком действия, например токен IAM в ElastiCache. Фича credential-provider добавляет credential_provider - колбэк, который возвращает пользователя и пароль для каждого AUTH или HELLO. Он также может заново входить на сервер с заданным интервалом, до истечения токена. Он важнее статических учётных данных.

Sentinel

src/main.rs

/// The primary `orders` as two sentinels report it, with the ACL user the deployment provides.
fn sentinel_broker(username: String, password: String) -> RedisBroker {
    RedisBroker::sentinel(
        "orders",
        ["sentinel-1.internal:26379", "sentinel-2.internal:26379"],
    )
    .credentials(username, password)
    .pool(8)
}

RedisBroker::sentinel называет ведущий узел, за которым следят узлы Sentinel, и сами узлы Sentinel. Клиент спрашивает у узлов Sentinel, где ведущий узел, и спрашивает снова после переключения при отказе. credentials выполняет аутентификацию на узлах с данными. Узлам Sentinel с собственным паролем нужен sentinel_credentials из фичи sentinel-auth.

Соединения и размер пула

Брокер держит пул соединений для публикаций, подтверждений и команд, которые отправляют обработчики. Его размер равен 4, если pool(n) не задаёт другой, а broker выше задаёт 8. Каждая подписка на поток и на список читает через собственное соединение, и каждая подписка Pub/Sub слушает своё, поэтому блокирующее чтение никогда не задерживает публикацию. Значит, сервис открывает пул и ещё по одному соединению на подписку. Рассчитайте лимит клиентов сервера с учётом этого.

Увеличьте размер пула, когда под нагрузкой публикации и подтверждения выстраиваются в очередь друг за другом. Подписка, смонтированная с .pipeline(), завершает весь полученный пакет за один обмен с сервером, и это снимает нагрузку с пула.

Переподключение

Каждое соединение, которое открывает брокер, переподключается после того, как сервер его разорвал, например при перезапуске или переключении при отказе. Первая попытка следует через 100 мс, каждая следующая ждёт вдвое дольше, до 10 с между попытками, а число попыток не ограничено. Команда, отправленная за это время, ждёт соединения.

Что делает подписка, когда соединение восстановлено:

• Подписка на поток продолжает чтение. Если сервер потерял группу потребителей, например при перезапуске без сохранения данных, подписка создаёт группу заново и читает поток с начала.
• Подписка Pub/Sub подписывается заново. Сообщения, опубликованные, пока она была отключена, теряются, потому что Pub/Sub их никогда не хранит.
• Шардированная подписка Pub/Sub в кластере не восстанавливается после того, как перераспределение слотов переносит слот её канала на другой узел. Сервер завершает такую подписку, не разрывая соединения, поэтому переподключения не происходит. Перезапустите сервис после перераспределения слотов.

Для другой политики переподключения или любой другой настройки, до которой не дотягиваются билдеры, постройте Pool из fred сами и оберните его через from_pool. Тогда собственные соединения подписок переподключаются по политике этого пула.

Остановка

Сервис останавливается по SIGINT или SIGTERM. Рантайм прекращает чтение, даёт обработчикам завершить доставки, которые у них на руках, и останавливает брокер: каждое заблокированное чтение сразу заканчивается, и каждое соединение закрывается. Запись, которую обработчик потока не успел подтвердить, остаётся неподтверждённой, и подписка с перехватом или reclaim забирает её после перезапуска.

Генерируемый документ

Документ AsyncAPI называет хост и порт сервера и никогда не содержит учётных данных. Брокер, построенный через from_pool, не сообщает хост, потому что пул его скрывает.