Перейти к основному содержимому
Нормативная версия документации - английская

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

Middleware и наблюдаемость

Приложение оборачивает каждого подписчика в два слоя: один пишет в журнал доставки и их исходы, другой, собственный, предупреждает о медленных обработчиках.

Layer оборачивает обработчик в другой обработчик. Слой выполняет код до тела обработчика, после него или вместо него и передаёт исход дальше. Слой приложения оборачивает всех подписчиков, которых монтирует приложение. Слой роутера оборачивает только подписчиков этого роутера. RustStream сам записывает события через tracing: запуск сервиса, подписку на каждую тему, остановку сервиса. TracingLayer добавляет событие для каждой доставки: delivery received на уровне DEBUG, затем её исход. Подтверждение (ack) записывается на уровне INFO, переобработка и отбрасывание - на уровне WARN. Фича logging выводит эти события, как показала глава 1. Вместо неё подходит любой другой подписчик tracing. Раздел Наблюдаемость описывает метрики Prometheus и экспорт в OpenTelemetry. Они подключаются как слои тем же способом.

Собственный слой магазина​

SlowHandlers замеряет время каждого обработчика и предупреждает, если обработчик работает дольше порога. Предупреждение называет подписку, имя которой слой читает из контекста доставки. У слоя нет формы на макросах: в обеих вкладках это один и тот же обычный тип. BlanketLayer позволяет приложению применять его к обработчикам, типы которых приложение скрывает.

observability.rs
/// Warns about every delivery whose handler ran longer than the threshold.
#[derive(Clone)]
pub(crate) struct SlowHandlers {
    threshold: Duration,
}

impl SlowHandlers {
    pub(crate) const fn over(threshold: Duration) -> Self {
        Self { threshold }
    }
}

/// The handler `SlowHandlers` makes of the one it wraps.
pub(crate) struct Timed<H> {
    inner: H,
    threshold: Duration,
}

impl<H> Layer<H> for SlowHandlers {
    type Handler = Timed<H>;

    fn layer(&self, inner: H) -> Timed<H> {
        Timed {
            inner,
            threshold: self.threshold,
        }
    }
}

impl<M, C, S, H> Handler<M, C, S> for Timed<H>
where
    M: Send + Sync,
    C: Send,
    S: Send + Sync,
    H: Handler<M, C, S>,
{
    async fn handle(&self, msg: &M, ctx: &mut Context<'_, C, S>) -> HandlerOutcome {
        let started = Instant::now();
        let outcome = self.inner.handle(msg, ctx).await;
        let took = started.elapsed();
        if took > self.threshold {
            tracing::warn!(subscription = ctx.name(), ?took, "slow handler");
        }
        outcome
    }
}

// The app wraps handlers whose types it hides, through this generic method.
impl BlanketLayer for SlowHandlers {
    fn apply<M, C, S, H>(&self, handler: H) -> impl Handler<M, C, S> + 'static
    where
        M: Send + Sync + 'static,
        C: Send + 'static,
        S: Send + Sync + 'static,
        H: Handler<M, C, S> + 'static,
    {
        self.layer(handler)
    }
}

Приложение​

service.rs собирает магазин. layer добавляет слой приложению. Первый добавленный слой - внешний. Роутер каждого модуля монтируется через include_router, и слои приложения оборачивают его подписчиков. on_startup создаёт разделяемое состояние из главы 7, а info() - описание из главы 9. Функция принимает брокер параметром, поэтому точка входа командной строки и HTTP-сервер из главы 10 собирают один и тот же магазин.

service.rs
/// The whole shop on `broker`. The command-line entry point and the HTTP host both build it here.
pub(crate) fn shop(
    broker: impl BrokerRegistration<Broker = MemoryBroker>,
) -> impl App<State = Shop> {
    RustStream::new(info())
        .on_startup(async move |()| Shop::open().await)
        .layer(TracingLayer::default())
        .layer(SlowHandlers::over(Duration::from_millis(250)))
        .with_broker(broker, |b| {
            b.include_router(orders::routes());
            b.include_router(payments::routes());
            b.include_router(shipping::routes());
            b.include_router(notifications::routes());
        })
}

Что показывает журнал​

Заказ, оформленный запросом с идентификатором req-1, оставляет по строке на каждый шаг:

INFO guide_http::orders order placed request=req-1 order=1 total_cents=14400
INFO ruststream::dispatch handler ack
INFO ruststream::dispatch handler ack
INFO ruststream::dispatch handler ack
INFO guide_http::notifications mail sent server=smtp://mail.internal:25 to="ada@example.com" text="We have your order 1."
INFO ruststream::dispatch handler ack
INFO ruststream::dispatch handler ack
INFO ruststream::dispatch handler ack
INFO guide_http::notifications mail sent server=smtp://mail.internal:25 to="ada@example.com" text="Order 1 is on its way: TRK-1-2."
INFO ruststream::dispatch handler ack

RUST_LOG настраивает, что выводится, например RUST_LOG=ruststream=debug,info.