Metadata-Version: 2.4
Name: fast-telemetry
Version: 2.0.0
Summary: Internal standardized metrics library
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: prometheus-client<1.0.0,>=0.21.0
Provides-Extra: web
Requires-Dist: fastapi<1.0.0,>=0.133.0; extra == "web"
Requires-Dist: prometheus-fastapi-instrumentator<9.0.0,>=8.1.0; extra == "web"
Provides-Extra: stream
Requires-Dist: faststream[prometheus]~=0.6.0; extra == "stream"
Provides-Extra: stream-rabbit
Requires-Dist: faststream[prometheus,rabbit]~=0.6.0; extra == "stream-rabbit"
Provides-Extra: stream-kafka
Requires-Dist: faststream[kafka,prometheus]~=0.6.0; extra == "stream-kafka"
Provides-Extra: stream-confluent
Requires-Dist: faststream[confluent,prometheus]~=0.6.0; extra == "stream-confluent"
Provides-Extra: stream-redis
Requires-Dist: faststream[prometheus,redis]~=0.6.0; extra == "stream-redis"
Provides-Extra: stream-nats
Requires-Dist: faststream[nats,prometheus]~=0.6.0; extra == "stream-nats"
Provides-Extra: grpc
Requires-Dist: grpcio<2.0.0,>=1.68.1; extra == "grpc"
Provides-Extra: grpc-codegen
Requires-Dist: protobuf<7.0.0,>=5.29.1; extra == "grpc-codegen"
Requires-Dist: grpcio-tools<2.0.0,>=1.68.1; extra == "grpc-codegen"
Provides-Extra: all
Requires-Dist: fast-telemetry[grpc,stream-confluent,stream-kafka,stream-nats,stream-rabbit,stream-redis,web]; extra == "all"

# 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`](docs/COOKBOOK.md).

## Установка

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

```bash
pip install fast-telemetry
```

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

```bash
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`:

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

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

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

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

```python
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, описанные ниже.

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

```python
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`:

```python
@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, те же метрики можно записать вручную:

```python
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 экземпляра:

```python
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` и настройте метрики после регистрации прикладных маршрутов либо до старта приложения:

```python
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:

```python
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-модель:

```python
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:

```python
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:

```python
import logging

from fast_telemetry import LOGGER_NAME, get_logger

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

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

```python
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:

```python
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:

```python
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`:

```python
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`.

## Метрики

| Metric | Type | Labels | Назначение |
|---|---|---|---|
| `<prefix>_app_info` | Gauge | `service`, `env`, `version` | Информация о приложении, значение `1` |
| `<prefix>_business_errors_total` | Counter | `error_type` | Ошибки бизнес-логики |
| `<prefix>_task_processing_seconds` | Histogram | `task_type` | Короткие внутренние задачи |
| `<prefix>_long_task_processing_seconds` | Histogram | `task_type` | Длительные задачи |
| `<prefix>_task_executions_total` | Counter | `task_type`, `result` | Результаты tracked tasks |
| `<prefix>_task_inprogress` | Gauge | `task_type` | Выполняющиеся tracked tasks |
| `<prefix>_http_*` | Counter/Histogram | HTTP labels | FastAPI RED-метрики |
| `<prefix>_grpc_*` | Counter/Histogram | RPC labels | gRPC client/server метрики |
| `<prefix>_faststream_*` | Counter/Histogram | Broker labels | FastStream метрики |

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

## Разработка

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

```bash
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.
