diff --git a/Makefile b/Makefile index c985b2a..5414029 100644 --- a/Makefile +++ b/Makefile @@ -9,7 +9,7 @@ test: docker compose -f deploy/docker-compose.test.yml up --build --abort-on-container-exit --exit-code-from test dev: - docker compose -f deploy/docker-compose.dev.yml up --build + docker compose -f deploy/docker-compose.dev.yml up --build -d cleanup: docker compose -f deploy/docker-compose.dev.yml down --remove-orphans diff --git a/bot/handlers/subscriptions.py b/bot/handlers/subscriptions.py index 5e232a1..f7ec64c 100644 --- a/bot/handlers/subscriptions.py +++ b/bot/handlers/subscriptions.py @@ -4,6 +4,7 @@ from aiogram.types import CallbackQuery, Message from sqlalchemy.ext.asyncio import AsyncSession +from bot.metrics import bot_handler_errors from bot.states.subscription import SearchForm from services.ps_store import GameInfo, RegionPrice from services.subscription import subscribe_to_game, unsubscribe_from_game @@ -21,6 +22,7 @@ async def on_subscribe(callback: CallbackQuery, state: FSMContext, session: Asyn index = int(callback.data.split(":", 1)[1]) if index >= len(entries): + bot_handler_errors.inc() await callback.message.answer("Game not found. Please search again.") return @@ -52,6 +54,7 @@ async def on_unsubscribe(callback: CallbackQuery, state: FSMContext, session: As index = int(callback.data.split(":", 1)[1]) if index >= len(entries): + bot_handler_errors.inc() await callback.message.answer("Game not found. Please search again.") return diff --git a/bot/main.py b/bot/main.py index b87a097..8e9a045 100644 --- a/bot/main.py +++ b/bot/main.py @@ -3,6 +3,7 @@ from aiogram import Bot, Dispatcher from aiogram.client.default import DefaultBotProperties from aiogram.enums import ParseMode +from prometheus_client import start_http_server from bot.handlers import router from bot.middlewares.db import DbSessionMiddleware @@ -13,6 +14,7 @@ async def main() -> None: + start_http_server(8000) bot = Bot( token=settings.BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML), diff --git a/bot/metrics.py b/bot/metrics.py new file mode 100644 index 0000000..b251a13 --- /dev/null +++ b/bot/metrics.py @@ -0,0 +1,36 @@ +from prometheus_client import Counter, Histogram + +ps_api_requests = Counter( + "ps_api_requests_total", + "Total PS Store API requests", + ["operation", "status"], +) +ps_api_none_results = Counter( + "ps_api_none_results_total", + "PS Store API calls that returned no usable result", + ["operation"], +) +ps_api_duration = Histogram( + "ps_api_request_duration_seconds", + "PS Store API request duration in seconds", + ["operation"], +) + +subscriptions_created = Counter( + "subscriptions_created_total", + "New subscriptions successfully created", +) +subscriptions_already_exists = Counter( + "subscriptions_already_exists_total", + "Subscribe attempts rejected because subscription already exists", +) + +bot_handler_errors = Counter( + "bot_messages_failed_total", + "Bot handler errors (stale state, missing entries, etc.)", +) + +region_sync_not_found = Counter( + "region_sync_games_not_found_total", + "Games not found in PS Store during region sync fallback search", +) diff --git a/config.py b/config.py index fdb1ad6..8edfc1e 100644 --- a/config.py +++ b/config.py @@ -10,7 +10,7 @@ class Settings(BaseSettings): SCHEDULER_INTERVAL_HOURS: int = 4 LOG_LEVEL: str = "INFO" - model_config = SettingsConfigDict(env_file="deploy/.env", env_file_encoding="utf-8") + model_config = SettingsConfigDict(env_file="deploy/.env", env_file_encoding="utf-8", extra="ignore") settings = Settings() diff --git a/deploy/.env.example b/deploy/.env.example index cee6d90..3f10ce0 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -1,4 +1,9 @@ +# Bot BOT_TOKEN=your_bot_token_here -DATABASE_URL=postgresql+asyncpg://pricestation:pricestation@localhost:5432/pricestation +DATABASE_URL=postgresql+asyncpg://pricestation:pricestation@db:5432/pricestation SCHEDULER_INTERVAL_HOURS=4 LOG_LEVEL=DEBUG + +# Alerts (Vector → Telegram) +ALERT_BOT_TOKEN=your_alert_bot_token_here +ALERT_CHAT_ID=your_telegram_chat_id_here diff --git a/deploy/docker-compose.prod.yml b/deploy/docker-compose.prod.yml index 2366e10..87fe5df 100644 --- a/deploy/docker-compose.prod.yml +++ b/deploy/docker-compose.prod.yml @@ -23,6 +23,25 @@ services: depends_on: db: condition: service_healthy + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000')"] + interval: 5s + timeout: 3s + retries: 5 + start_period: 10s + restart: unless-stopped + + alerts: + container_name: pricestation-alerts + image: timberio/vector:0.42.0-alpine + command: ["--config", "/etc/vector/vector.toml"] + volumes: + - ./vector/vector.toml:/etc/vector/vector.toml:ro + - /var/run/docker.sock:/var/run/docker.sock:ro + env_file: .env + depends_on: + bot: + condition: service_healthy restart: unless-stopped volumes: diff --git a/deploy/vector/README.md b/deploy/vector/README.md new file mode 100644 index 0000000..25d438d --- /dev/null +++ b/deploy/vector/README.md @@ -0,0 +1,52 @@ +# Vector alerting + +[Vector](https://vector.dev) - высокопроизводительный pipeline для сбора и маршрутизации логов и метрик. + +Сбор логов и метрик бота с отправкой алертов в Telegram. + +## Sources + +Точки входа - откуда Vector получает данные. Каждый source непрерывно собирает события и передаёт их в transforms. + +| ID | Тип | Описание | +|---------------|---------------------|--------------------------------------------------------| +| `bot_logs` | `docker_logs` | Логи контейнера `pricestation-bot` | +| `bot_metrics` | `prometheus_scrape` | Метрики с `http://pricestation-bot:8000` каждые 60 сек | + +## Transforms + +Промежуточная обработка событий: фильтрация, троттлинг и форматирование перед отправкой. Каждый transform принимает события от одного или нескольких источников и передаёт результат дальше по цепочке. + +| ID | Описание | +|--------------------------|----------------------------------------------------------------| +| `filter_log_errors` | Фильтр логов по `ERROR\|Exception\|Traceback` | +| `throttle_log_errors` | Не более 1 лог-алерта за 5 минут | +| `format_log_alert` | Формирует JSON-тело для Telegram | +| `filter_error_metrics` | Фильтр по отслеживаемым счётчикам | +| `detect_increases` | Lua: дельта между скрейпами, эмитит событие при росте счётчика | +| `throttle_metric_alerts` | Не более 1 алерта на метрику за 5 минут | +| `format_metric_alert` | Формирует JSON-тело для Telegram | + +## Отслеживаемые метрики + +| Метрика | Условие алерта | +|-------------------------------------|-------------------------------------------| +| `ps_api_requests_total` | Рост счётчика со статусом `4xx` или `5xx` | +| `ps_api_none_results_total` | Любой рост | +| `bot_messages_failed_total` | Любой рост | +| `region_sync_games_not_found_total` | Любой рост | + +## Sinks + +Точки назначения - куда Vector отправляет финальные события. В отличие от transforms, sinks не передают события дальше, а записывают или отправляют их во внешние системы. + +| ID | Тип | Описание | +|------------|--------|-----------------------------------------| +| `telegram` | `http` | POST в Telegram Bot API (`sendMessage`) | + +## Переменные окружения + +| Переменная | Описание | +|-------------------|--------------------------------------------| +| `ALERT_BOT_TOKEN` | Токен Telegram-бота для отправки алертов | +| `ALERT_CHAT_ID` | ID чата/пользователя для получения алертов | diff --git a/deploy/vector/vector.toml b/deploy/vector/vector.toml new file mode 100644 index 0000000..678945a --- /dev/null +++ b/deploy/vector/vector.toml @@ -0,0 +1,131 @@ +# Sources + +[sources.bot_logs] +type = "docker_logs" +include_containers = ["pricestation-bot"] + +[sources.bot_metrics] +type = "prometheus_scrape" +endpoints = ["http://pricestation-bot:8000"] +scrape_interval_secs = 60 + +# Log-based alerts + +[transforms.filter_log_errors] +type = "filter" +inputs = ["bot_logs"] +condition.type = "vrl" +condition.source = "match!(.message, r'ERROR|Exception|Traceback')" + +[transforms.throttle_log_errors] +type = "throttle" +inputs = ["filter_log_errors"] +threshold = 1 +window_secs = 300 + +[transforms.format_log_alert] +type = "remap" +inputs = ["throttle_log_errors"] +source = ''' + _msg = string!(.message) + .message = encode_json({ + "chat_id": to_int!("${ALERT_CHAT_ID}"), + "text": "🚨 PriceStation Log Alert\n
" + _msg + "", + "parse_mode": "HTML" + }) +''' + +# Metrics-based alerts + +[transforms.filter_error_metrics] +type = "filter" +inputs = ["bot_metrics"] +condition.type = "vrl" +condition.source = '.name == "ps_api_requests_total" || .name == "ps_api_none_results_total" || .name == "bot_messages_failed_total" || .name == "region_sync_games_not_found_total"' + +[transforms.detect_increases] +type = "lua" +inputs = ["filter_error_metrics"] +version = "2" +hooks.init = "init" +hooks.process = "process" +source = """ +function init() + _prev = {} +end + +function process(event, emit) + local name = event.metric.name + local key = name + local tags_str = "" + + if event.metric.tags then + for k, v in pairs(event.metric.tags) do + key = key .. "|" .. k .. "=" .. v + tags_str = tags_str .. " " .. k .. "=" .. v + end + end + + -- For ps_api_requests_total, only alert on 4xx/5xx status + if name == "ps_api_requests_total" then + local status = (event.metric.tags and event.metric.tags.status) or "" + local first = string.sub(status, 1, 1) + if first ~= "4" and first ~= "5" then + _prev[key] = event.metric.counter and event.metric.counter.value or 0 + return + end + end + + local current_val = (event.metric.counter and event.metric.counter.value) or 0 + local prev_val = _prev[key] + _prev[key] = current_val + + if prev_val ~= nil and current_val > prev_val then + emit({ + log = { + metric_name = name, + tags_str = tags_str, + delta = tostring(math.floor(current_val - prev_val)), + } + }) + end +end +""" + +[transforms.throttle_metric_alerts] +type = "throttle" +inputs = ["detect_increases"] +threshold = 1 +window_secs = 300 +key_field = "metric_name" + +[transforms.format_metric_alert] +type = "remap" +inputs = ["throttle_metric_alerts"] +source = ''' + _name = string!(.metric_name) + _tags = string!(.tags_str) + _delta = string!(.delta) + .message = encode_json({ + "chat_id": to_int!("${ALERT_CHAT_ID}"), + "text": "🚨 PriceStation Metric Alert\n
" + _name + _tags + "\n+" + _delta + " за последние 60 сек",
+ "parse_mode": "HTML"
+ })
+'''
+
+# Sink
+
+[sinks.telegram]
+type = "http"
+inputs = ["format_log_alert", "format_metric_alert"]
+uri = "https://api.telegram.org/bot${ALERT_BOT_TOKEN}/sendMessage"
+method = "post"
+encoding.codec = "raw_message"
+
+[sinks.telegram.batch]
+max_events = 1
+timeout_secs = 1
+
+[sinks.telegram.request]
+retry_attempts = 3
+headers.Content-Type = "application/json"
diff --git a/requirements.txt b/requirements.txt
index 276b4c3..3fe3f48 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -9,4 +9,5 @@ pytest==8.3.5
pytest-asyncio==0.25.3
pytest-cov==6.1.0
pytest-mock==3.14.0
+prometheus-client==0.21.1
ruff==0.11.8
diff --git a/services/ps_store.py b/services/ps_store.py
index 82e031a..e2aaeee 100644
--- a/services/ps_store.py
+++ b/services/ps_store.py
@@ -1,11 +1,13 @@
import json
import logging
import re
+import time
from dataclasses import dataclass, field
from urllib.parse import urlencode
import aiohttp
+from bot.metrics import ps_api_duration, ps_api_none_results, ps_api_requests
from services.currency import PS_CURRENCY_MAP, PS_ISO_TO_SYMBOL
logger = logging.getLogger(__name__)
@@ -287,17 +289,25 @@ async def search_games(
headers = _gql_headers(region, "https://store.playstation.com/")
words = [w.lower() for w in query.split() if w]
+ _t0 = time.monotonic()
async with aiohttp.ClientSession() as session:
async with session.get(f"{_GQL_URL}?{params}", headers=headers) as resp:
+ _status = resp.status
if resp.status != 200:
level = logging.WARNING if resp.status in _WARN_STATUSES else logging.ERROR
logger.log(level, "search_games: HTTP %d [query=%r region=%s]", resp.status, query, region)
+ ps_api_requests.labels(operation="search", status=str(_status)).inc()
+ ps_api_duration.labels(operation="search").observe(time.monotonic() - _t0)
+ ps_api_none_results.labels(operation="search").inc()
return []
data = await resp.json(content_type=None)
+ ps_api_requests.labels(operation="search", status="200").inc()
+ ps_api_duration.labels(operation="search").observe(time.monotonic() - _t0)
page = (data.get("data") or {}).get("universalSearch")
if not page:
logger.warning("search_games: no universalSearch data [query=%r region=%s]", query, region)
+ ps_api_none_results.labels(operation="search").inc()
return []
results: list[tuple[GameInfo, RegionPrice]] = []
@@ -336,20 +346,28 @@ async def get_game_info(ps_id: str, region: str = "en-us") -> tuple[GameInfo, Re
"extensions": json.dumps({"persistedQuery": {"version": 1, "sha256Hash": _GQL_UPSELL_HASH}}),
})
+ _t0 = time.monotonic()
async with aiohttp.ClientSession() as session:
async with session.get(
f"{_GQL_URL}?{params}",
headers=_gql_headers(region, f"https://store.playstation.com/{region}/product/{ps_id}/"),
) as resp:
+ _status = resp.status
if resp.status != 200:
level = logging.WARNING if resp.status in _WARN_STATUSES else logging.ERROR
logger.log(level, "get_game_info: HTTP %d [ps_id=%s region=%s]", resp.status, ps_id, region)
+ ps_api_requests.labels(operation="get_game_info", status=str(_status)).inc()
+ ps_api_duration.labels(operation="get_game_info").observe(time.monotonic() - _t0)
+ ps_api_none_results.labels(operation="get_game_info").inc()
return None
data = await resp.json(content_type=None)
+ ps_api_requests.labels(operation="get_game_info", status="200").inc()
+ ps_api_duration.labels(operation="get_game_info").observe(time.monotonic() - _t0)
retrieve = (data.get("data") or {}).get("productRetrieve")
if not retrieve:
logger.warning("get_game_info: product not found [ps_id=%s region=%s]", ps_id, region)
+ ps_api_none_results.labels(operation="get_game_info").inc()
return None
products = (retrieve.get("concept") or {}).get("products") or []
@@ -359,6 +377,7 @@ async def get_game_info(ps_id: str, region: str = "en-us") -> tuple[GameInfo, Re
"get_game_info: product not in concept.products [ps_id=%s region=%s]",
ps_id, region,
)
+ ps_api_none_results.labels(operation="get_game_info").inc()
return None
webctas = product.get("webctas") or []
diff --git a/services/subscription.py b/services/subscription.py
index 5ed26a9..020b4c9 100644
--- a/services/subscription.py
+++ b/services/subscription.py
@@ -6,6 +6,7 @@
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased
+from bot.metrics import region_sync_not_found, subscriptions_already_exists, subscriptions_created
from db.models.game import Game
from db.models.game_region import GameRegion
from db.models.region import Region
@@ -96,6 +97,7 @@ async def subscribe_to_game(
"subscribed telegram_id=%d to new game %r regions=%s",
user.telegram_id, game_info.title, region_codes,
)
+ subscriptions_created.inc()
return True
# Game already exists
@@ -103,6 +105,7 @@ async def subscribe_to_game(
existing_sub: Subscription | None = rows[0][3]
if existing_sub is not None:
+ subscriptions_already_exists.inc()
return False
# Prefer ASCII title (same rule as in search merge: localized prefixes like "Набір" lose to ASCII)
@@ -151,6 +154,7 @@ async def subscribe_to_game(
"subscribed telegram_id=%d to existing game_id=%d %r",
user.telegram_id, game.id, game.title,
)
+ subscriptions_created.inc()
return True
@@ -270,6 +274,7 @@ async def sync_subscriptions_for_new_region(
"sync fallback: no match for game_id=%d normalized=%r region=%s",
game_id, normalized_title, region.code,
)
+ region_sync_not_found.inc()
continue
region_prices.append((game_id, rp))