fast-telemetry

Единый Prometheus toolkit для Python-сервисов на FastAPI, FastStream и gRPC, а также для фоновых workers и batch jobs.

Библиотека предоставляет общий registry и namespace, стандартную метрику информации о приложении, бизнес-метрики, автоматическую инструментацию transport-слоя и готовые способы экспорта через HTTP или Pushgateway.

Возможности

• PrometheusExporter — минимальный registry, prefix и factory API без предустановленных прикладных метрик.
• PrometheusMetrics — готовый набор: app_info, счётчик бизнес-ошибок и histogram времени задач.
• Декораторы и context managers для sync/async функций: measure_task, track_exception, timer.
• Составной track_task: duration, in-progress и результат success|error|cancelled.
• Factory API для пользовательских Counter, Gauge, Histogram и Summary в том же prefix/registry.
• FastAPI RED-метрики и endpoint /metrics, включая mounted AsgiFastStream без monkey patching.
• FastStream middleware для RabbitMQ, Kafka, Confluent, Redis и NATS.
• Sync и async gRPC interceptors для client/server, unary и streaming RPC.
• Worker HTTP endpoint и отправка batch-метрик в Pushgateway.
• Типизированный PEP 561 package (py.typed).

При использовании PrometheusMetrics значения service, env и version публикуются в fasttelemetry_app_info. Минимальный PrometheusExporter не создаёт app_info и другие стандартные collectors. Метки target-level лучше назначать в Prometheus через scrape/relabel, чтобы не дублировать их в каждой метрике.

Практическое руководство по архитектуре, DI, наследованию, пользовательским метрикам и production-эксплуатации: docs/COOKBOOK.md.

Установка

Базовые метрики и worker-интеграция:

pip install fast-telemetry

Устанавливайте только необходимые transport dependencies:

pip install "fast-telemetry[web]"
pip install "fast-telemetry[stream-rabbit]"
pip install "fast-telemetry[stream-kafka]"
pip install "fast-telemetry[stream-confluent]"
pip install "fast-telemetry[stream-redis]"
pip install "fast-telemetry[stream-nats]"
pip install "fast-telemetry[grpc]"

grpc-codegen нужен только окружению, которое генерирует Python-код из .proto:

pip install "fast-telemetry[grpc,grpc-codegen]"

Все runtime-интеграции можно установить через fast-telemetry[all].

Только пользовательские метрики

Если готовые app_info, error и task collectors не нужны, используйте минимальный exporter. До объявления пользовательских collectors его registry пуст, но его можно передавать во все transport integrations:

from fast_telemetry import PrometheusExporter

metrics = PrometheusExporter(service_name="payment-api", prefix="payments")
created = metrics.counter(
    "created_total",
    "Number of created payments",
    labelnames=("provider",),
)
created.labels(provider="bank").inc()

PrometheusMetrics наследует этот factory API и добавляет стандартный набор и wrappers, описанные ниже.

Базовые метрики

from fast_telemetry import PrometheusMetrics

metrics = PrometheusMetrics(
    service_name="payment-api",
    version="1.4.0",
    env="production",
    prefix="payments",
    fast_buckets=(0.01, 0.05, 0.1, 0.5, 1.0),
    long_buckets=(5.0, 30.0, 60.0, 300.0),
)

metrics.inc_error("validation")

with metrics.timer("database_query"):
    load_payment()


@metrics.measure_task("reconciliation", long_task=True)
def reconcile() -> None:
    process_payments()


@metrics.track_exception(exclude=[KeyError])
async def charge() -> None:
    await payment_gateway.charge()

Для операций, где нужны duration, число завершений и concurrency одновременно, используйте track_task:

@metrics.track_task("charge_payment")
async def charge() -> None:
    await payment_gateway.charge()


async with metrics.track_task("reconciliation", long_task=True):
    await reconcile()

Он публикует существующую histogram времени, task_executions_total{task_type,result} и task_inprogress{task_type}. Исключения не подавляются; отменённая async-задача получает result="cancelled".

Если жизненным циклом управляет внешний scheduler или callback API, те же метрики можно записать вручную:

metrics.record_task_started("provider_sync")
try:
    synchronize()
except Exception:
    metrics.record_task_finished("provider_sync", duration=1.2, result="error")
    raise
else:
    metrics.record_task_finished("provider_sync", duration=1.2, result="success")

Для обычного прикладного кода предпочтительнее track_task: он самостоятельно измеряет duration и гарантирует согласованное обновление gauge и counter.

Пользовательские метрики

Factory-методы автоматически используют prefix и registry экземпляра:

provider_failures = metrics.counter(
    "provider_failures_total",
    "Failed provider calls",
    labelnames=("provider", "reason"),
)
active_payments = metrics.gauge("active_payments", "Payments currently being processed")
provider_latency = metrics.histogram(
    "provider_request_seconds",
    "Provider latency",
    labelnames=("provider",),
    buckets=(0.05, 0.1, 0.25, 0.5, 1.0, 2.0),
)
payload_size = metrics.summary("payload_bytes", "Payload size", unit="bytes")

provider_failures.labels(provider="bank", reason="timeout").inc()

Повторный вызов factory с тем же именем и конфигурацией возвращает тот же collector. Конфликт типа, labels, buckets или документации завершается ValueError. Label names задаются при создании: не используйте высококардинальные значения вроде user_id, request_id, URL или текста исключения.

Если env и version не переданы, используются APP_ENV (dev) и APP_VERSION (unknown). Prefix должен соответствовать правилам имён Prometheus.

FastAPI

Установите extra web и настройте метрики после регистрации прикладных маршрутов либо до старта приложения:

from fastapi import FastAPI

from fast_telemetry import PrometheusMetrics
from fast_telemetry.integrations.fastapi import FastAPIMetricsConfig, setup_fastapi_metrics

app = FastAPI(title="payment-api", version="1.4.0")
metrics = PrometheusMetrics("payment-api", version=app.version, env="production")


@app.get("/payments/{payment_id}")
async def get_payment(payment_id: str) -> dict[str, str]:
    return {"payment_id": payment_id}


setup_fastapi_metrics(
    app,
    metrics,
    config=FastAPIMetricsConfig(
        path="/metrics",
        excluded_routes=("/health",),
        should_group_status_codes=True,
        should_instrument_requests_inprogress=True,
        should_gzip=True,
    ),
)

Повторный вызов с тем же приложением, exporter и path безопасен и возвращает False. Конфликт с уже занятым path завершается ValueError. Mounted FastStream tuple-routes разрешаются локальным middleware библиотеки; глобальный monkey patch prometheus_fastapi_instrumentator не требуется.

FastStream

Выберите extra для своего брокера. Пример RabbitMQ:

from faststream.asgi import AsgiFastStream, make_ping_asgi
from faststream.rabbit import RabbitBroker

from fast_telemetry import PrometheusMetrics
from fast_telemetry.integrations.faststream import setup_faststream_metrics

broker = RabbitBroker("amqp://guest:guest@rabbitmq:5672/")


@broker.subscriber("payments")
async def process_payment(message: dict[str, object]) -> None: ...


app = AsgiFastStream(
    broker,
    asgi_routes=[("/health", make_ping_asgi(broker))],
    asyncapi_path="/docs",
)
metrics = PrometheusMetrics("payment-consumer", version="1.4.0", env="production")
setup_faststream_metrics(app, metrics, path="/metrics")

Функция автоматически выбирает Prometheus middleware по типу broker. Пользовательский broker можно добавить через register_custom_broker. Повторная настройка той же пары exporter/path возвращает False и не дублирует middleware.

Workers и Pushgateway

Для long-running worker используйте pull-модель:

from fast_telemetry.integrations.worker import WorkerMetrics

metrics = WorkerMetrics("thumbnail-worker", version="1.4.0", env="production")
metrics.start_server(port=8000)

Для batch job метрики отправляются при выходе из sync или async context manager:

metrics = WorkerMetrics("nightly-reconciliation", instance_id="nightly")
tracker = metrics.track_job(
    "http://pushgateway:9091",
    grouping_key={"instance": metrics.instance_id},
    raise_on_error=True,
)

with tracker:
    reconcile()

# Удалите группу, когда её жизненный цикл действительно завершён:
tracker.delete()

По умолчанию ошибки Pushgateway логируются и не прерывают job. raise_on_error=True включает fail-fast. Стандартный instance_id равен hostname и не создаёт новую группу при каждом изменении PID; его можно задать явно.

Логирование

Библиотека пишет диагностические сообщения через стандартный logging в logger fast_telemetry и не настраивает уровни, handlers или root logger приложения. Стабильное имя и сам logger доступны из публичного API:

import logging

from fast_telemetry import LOGGER_NAME, get_logger

logging.getLogger(LOGGER_NAME).setLevel(logging.INFO)
library_logger = get_logger()

Приложение полностью управляет маршрутизацией сообщений. Например, для передачи их в Loguru установите handler на logger библиотеки:

import logging

from loguru import logger

from fast_telemetry import LOGGER_NAME


class LoguruHandler(logging.Handler):
    def emit(self, record: logging.LogRecord) -> None:
        logger.opt(exception=record.exc_info).log(record.levelname, record.getMessage())


logging.getLogger(LOGGER_NAME).handlers = [LoguruHandler()]

URL Pushgateway очищается от userinfo перед записью ошибки, поэтому пароль из https://user:password@host не попадает в сообщения библиотеки.

gRPC

Sync server:

from concurrent import futures

import grpc

from fast_telemetry import PrometheusMetrics
from fast_telemetry.integrations.grpc import create_server_metrics_interceptor

metrics = PrometheusMetrics("payment-grpc")
server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[create_server_metrics_interceptor(metrics)],
)

Sync client:

from fast_telemetry.integrations.grpc import create_client_metrics_interceptor

base_channel = grpc.insecure_channel("payment-grpc:50051")
channel = grpc.intercept_channel(base_channel, create_client_metrics_interceptor(metrics))

Async server/client используют фабрики из fast_telemetry.integrations.grpc.aio:

from fast_telemetry.integrations.grpc.aio import (
    create_client_metrics_interceptor,
    create_server_metrics_interceptor,
)

server = grpc.aio.server(interceptors=[create_server_metrics_interceptor(metrics)])
channel = grpc.aio.insecure_channel(
    "payment-grpc:50051",
    interceptors=create_client_metrics_interceptor(metrics),
)

Registry можно экспортировать стандартным prometheus_client.start_http_server.

Метрики

Готовые dashboards находятся в каталоге grafana. Полный Docker Compose пример FastAPI, RabbitMQ, FastStream, worker, sync/async gRPC, Prometheus, Pushgateway и Grafana — в examples/app_example.

Разработка

Проект использует uv и lock-файл:

uv sync --locked --all-extras --dev
uv run ruff check .
uv run ruff format --check .
uv run mypy
uv run pytest -q --cov=fast_telemetry
uv build

GitLab CI/CD запускает проверки для Python 3.11–3.14. По Git-тегу pipeline собирает и валидирует wheel/sdist, затем публикует дистрибутивы в настроенный приватный package registry.