Add vector alerting bot

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

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
 

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),

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",
+)

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()

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

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:

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 чата/пользователя для получения алертов |

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": "<b>🚨 PriceStation Log Alert</b>\n<pre>" + _msg + "</pre>",
+    "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": "<b>🚨 PriceStation Metric Alert</b>\n<code>" + _name + _tags + "</code>\n+<b>" + _delta + "</b> за последние 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"

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

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 []