From fad2bd3353805c72925dfc98816e4a0d96b4612c Mon Sep 17 00:00:00 2001 From: Denozordec Date: Mon, 1 Jun 2026 13:43:33 +0700 Subject: [PATCH] feat(db): implement PostgreSQL monitoring and maintenance features Added PostgreSQL monitoring and maintenance capabilities to the API, including new endpoints for instance-level metrics, maintenance operations, and job scheduling. Updated the HTTP API to support PostgreSQL monitoring routes and integrated a background scheduler for metrics collection. Enhanced the CLI with database commands for maintenance tasks. Updated documentation to reflect these changes. --- .../технический_аудит_evobgp_60e04155.plan.md | 408 +++++++++++ cmd/evobgp-all/main.go | 5 + cmd/evobgp-api/main.go | 5 + docs/access.md | 12 + docs/db-diagnostics.md | 15 + docs/openapi.yaml | 332 +++++++++ internal/dbcli/dbcli.go | 209 ++++++ internal/httpapi/bootstrap.go | 2 +- internal/httpapi/routes.go | 2 + .../httpapi/routes_postgres_maintenance.go | 203 ++++++ .../httpapi/routes_postgres_monitoring.go | 130 ++++ internal/httpapi/routes_postgres_test.go | 21 + internal/httpapi/server.go | 14 + internal/jobs/postgres_worker.go | 169 +++++ internal/jobs/worker.go | 45 +- internal/pgmonitor/cache.go | 37 + internal/pgmonitor/correlation.go | 83 +++ internal/pgmonitor/maintenance_audit.go | 154 ++++ internal/pgmonitor/queries.go | 294 ++++++++ internal/pgmonitor/recommendations.go | 104 +++ internal/pgmonitor/scheduler.go | 59 ++ internal/pgmonitor/service.go | 90 +++ internal/pgmonitor/service_test.go | 33 + internal/pgmonitor/snapshot.go | 158 +++++ internal/pgmonitor/types.go | 148 ++++ .../000023_pg_stat_statements.down.sql | 1 + .../postgres/000023_pg_stat_statements.up.sql | 2 + .../000024_postgres_monitoring.down.sql | 2 + .../000024_postgres_monitoring.up.sql | 29 + .../sqlite/000023_pg_stat_statements.down.sql | 1 + .../sqlite/000023_pg_stat_statements.up.sql | 1 + .../000024_postgres_monitoring.down.sql | 2 + .../sqlite/000024_postgres_monitoring.up.sql | 20 + .../monitoring/MonitoringPostgresTab.svelte | 500 +++++++++++++ web/src/lib/monitoring/postgres.ts | 116 +++ web/src/routes/monitoring/+page.svelte | 658 +++++++++--------- 36 files changed, 3742 insertions(+), 322 deletions(-) create mode 100644 .cursor/plans/технический_аудит_evobgp_60e04155.plan.md create mode 100644 internal/dbcli/dbcli.go create mode 100644 internal/httpapi/routes_postgres_maintenance.go create mode 100644 internal/httpapi/routes_postgres_monitoring.go create mode 100644 internal/httpapi/routes_postgres_test.go create mode 100644 internal/jobs/postgres_worker.go create mode 100644 internal/pgmonitor/cache.go create mode 100644 internal/pgmonitor/correlation.go create mode 100644 internal/pgmonitor/maintenance_audit.go create mode 100644 internal/pgmonitor/queries.go create mode 100644 internal/pgmonitor/recommendations.go create mode 100644 internal/pgmonitor/scheduler.go create mode 100644 internal/pgmonitor/service.go create mode 100644 internal/pgmonitor/service_test.go create mode 100644 internal/pgmonitor/snapshot.go create mode 100644 internal/pgmonitor/types.go create mode 100644 migrations/postgres/000023_pg_stat_statements.down.sql create mode 100644 migrations/postgres/000023_pg_stat_statements.up.sql create mode 100644 migrations/postgres/000024_postgres_monitoring.down.sql create mode 100644 migrations/postgres/000024_postgres_monitoring.up.sql create mode 100644 migrations/sqlite/000023_pg_stat_statements.down.sql create mode 100644 migrations/sqlite/000023_pg_stat_statements.up.sql create mode 100644 migrations/sqlite/000024_postgres_monitoring.down.sql create mode 100644 migrations/sqlite/000024_postgres_monitoring.up.sql create mode 100644 web/src/lib/components/monitoring/MonitoringPostgresTab.svelte create mode 100644 web/src/lib/monitoring/postgres.ts diff --git a/.cursor/plans/технический_аудит_evobgp_60e04155.plan.md b/.cursor/plans/технический_аудит_evobgp_60e04155.plan.md new file mode 100644 index 0000000..31e6ca8 --- /dev/null +++ b/.cursor/plans/технический_аудит_evobgp_60e04155.plan.md @@ -0,0 +1,408 @@ +--- +name: Технический аудит EvoBGP +overview: "Полный технический аудит EvoBGP для production-сценария (10+ клиентов, нестабильная сеть). Архитектура — hybrid control plane; сильные стороны: stale fallback, CDN/RIPEstat resilience, Ed25519 bundles. Критичные риски: in-process jobs, DoH без retry, misconfiguration demo-seed, отсутствие HA API." +todos: + - id: quick-ops-checklist + content: Применить production-checklist (SEED_DEMO=0, BUNDLE_SEED_HEX, DB/JOB/CONCURRENCY tuning, TLS) + status: pending + - id: fix-doh-retry + content: Добавить DoWithRetry для DoH в internal/pipeline/refresh.go + status: pending + - id: fix-job-meta-race + content: Исправить чтение j.Meta в worker.go через Snapshot() или locked accessor + status: pending + - id: cdn-preview-resilience + content: Перевести CDN preview на upstreamHTTPDo в routes_crud.go + status: pending + - id: partial-cdn-failure + content: "Partial CDN source failure: skip/degrade вместо fail всего модуля" + status: pending + - id: graceful-shutdown + content: Cancel/drain jobs при SIGTERM в cmd/evobgp-api и evobgp-all + status: pending + - id: ha-job-queue + content: "Roadmap: распределённая очередь jobs (PG claim или NATS) для HA API" + status: pending +isProject: false +--- + +# Технический аудит EvoBGP + +## Executive summary + +EvoBGP — **hybrid control plane**: один процесс [`evobgp-all`](cmd/evobgp-all/main.go) (monolith) или **reference Compose** с разделёнными воркерами ([`docs/architecture.md`](docs/architecture.md)). Data plane (BIRD + agent) отделён от control plane (API + PostgreSQL + jobs). + +**Сильные стороны для нестабильной сети:** +- Stale snapshot fallback по умолчанию (`EVOBGP_STALE_ON_UPSTREAM_ERROR=1`) — [`internal/pipeline/collect_stale.go`](internal/pipeline/collect_stale.go) +- CDN/RIPEstat: retry (3×) + per-host circuit breaker — [`internal/httpclient/httpclient.go`](internal/httpclient/httpclient.go), [`circuit.go`](internal/httpclient/circuit.go) +- ETag conditional GET, ASN TTL-кэш, parallel collect с cap +- Подписанные бандлы Ed25519, verify перед apply + +**Главные риски для 10+ клиентов:** +1. `jobs.Registry` — **in-memory, только в процессе API** (ARCH-04) +2. DoH — **без retry/breaker** (критично при блокировках провайдеров) +3. Один failed CDN source **без stale cache валит весь модуль** +4. Production misconfiguration: `Bearer dev`, HTTP API, ephemeral bundle key +5. Data race на `Job.Meta` и alias pointers в `store.Memory` + +--- + +## 1. Архитектура + +### Стиль + +```mermaid +flowchart TB + subgraph hybrid [Hybrid deployment] + All[evobgp_all monolith] + Split[evobgp_api + workers] + end + subgraph cp [Control plane] + API[HTTP API] + Jobs[jobs.Registry in-process] + PG[(PostgreSQL)] + end + subgraph dp [Data plane per speaker] + Agent[evobgp_agent] + BIRD[BIRD2] + NodeCLI[evobgp_node] + end + All --> API + Split --> API + API --> Jobs + API --> PG + NodeCLI --> API + Agent --> API + Agent --> BIRD +``` + +| Профиль | Стиль | Когда | +|---------|-------|-------| +| `microvps` / `evobgp-all` | Monolith | 1 VPS, shared Registry | +| reference Compose | Microservices-lite | API + scheduler/ingest/render/deploy | +| Remote speakers | Edge agents | Panel→Node dispatch | + +### Узкие места (bottlenecks) + +| # | Bottleneck | Где | Impact | +|---|------------|-----|--------| +| B1 | **In-process job queue** | [`internal/jobs/job.go:175-177`](internal/jobs/job.go) | HA API невозможен без потери/дублирования jobs; scheduler без `EVOBGP_CONTROL_PLANE_URL` создаёт **отдельный Registry** — [`cmd/evobgp-scheduler/main.go:58-60`](cmd/evobgp-scheduler/main.go) | +| B2 | **Module refresh = sync upstream fan-out** | [`internal/pipeline/collect_parallel.go`](internal/pipeline/collect_parallel.go) | До `EVOBGP_COLLECT_CONCURRENCY` (8 default, max 32) параллельных HTTP; worst case ~45s × retries на источник | +| B3 | **Default job concurrency = 8** | [`internal/jobs/job.go:264-268`](internal/jobs/job.go) | При burst refresh 10+ tenants — очередь растёт, goroutine блокируются на sem | +| B4 | **PostgreSQL pool default** | [`internal/db/open.go:28-38`](internal/db/open.go) | pgx default ~4 conns; при `JOB_MAX=16` + HTTP — contention без `EVOBGP_DB_MAX_CONNS=25` | +| B5 | **Live endpoints fan-out** | [`internal/httpapi/peers_live.go`](internal/httpapi/peers_live.go) | N goroutines × N speakers, 12s timeout каждый | +| B6 | **Broker — заглушка** | [`internal/broker`](internal/broker) | NATS URL логируется, очередь не распределена | + +### Масштабируемость + +- **Вертикальная:** хорошо до ~10–20 tenants при `evobgp-all` + tuning ([`docs/production-checklist.md`](docs/production-checklist.md)) +- **Горизонтальная API:** **не поддерживается** — два `evobgp-api` = два независимых Registry; `job_audit` в PG — audit only, не очередь исполнения +- **Workers (ingest/render/deploy):** координируются через **общую БД**, не через jobs — OK для prefetch/drift + +### Отказоустойчивость + +| Сценарий | Поведение | Оценка | +|----------|-----------|--------| +| CDN/RIPEstat недоступен | Stale snapshot + circuit breaker | **Хорошо** (если был prior snapshot) | +| DoH недоступен | Fail модуля или stale domain snapshot | **Средне** (нет HTTP retry) | +| API restart mid-job | Job теряется из Registry; audit может быть inconsistent | **Плохо** | +| PG недоступен | API `/v1/ready` → 503 | **OK** | +| Agent unreachable | Deploy job succeed, drift в `evobgp-deploy` | **Частичный fail** (by design) | + +**Рекомендация:** для 10+ клиентов — **`evobgp-all` на каждом CP** или один CP + tuning; HA API требует **распределённой очереди** (NATS/Redis + worker pool) — задокументировано как future work. + +--- + +## 2. Анализ кода + +### Антипаттерны + +| ID | Проблема | Файл | Критичность | +|----|----------|------|-------------| +| A1 | **Concurrent map read/write** — worker читает `j.Meta` без lock, handler пишет через `mergeMeta`/`Snapshot` | [`worker.go:108,263,379`](internal/jobs/worker.go), [`job.go:103-111`](internal/jobs/job.go) | **high** | +| A2 | **Escape internal pointers** из Memory store | [`store/memory.go:416-475`](internal/store/memory.go) | **high** (tests/dev); **low** (prod PG) | +| A3 | **Fire-and-forget goroutine** на каждый auth | [`auth.go:79-81`](internal/httpapi/auth.go) | **medium** | +| A4 | **Silent error swallow** в prefetch | [`internal/ingest/run.go`](internal/ingest/run.go), `prefetch.go` | **medium** | +| A5 | **Bypass resilience layer** — CDN preview прямой `Do` | [`routes_crud.go:267`](internal/httpapi/routes_crud.go) | **medium** | +| A6 | **`EVOBGP_DEV_INSECURE` — dead code** | compose + [`server.go`](internal/httpapi/server.go) | **low** (misleading ops) | +| A7 | **Unused Registry** в ingest/render/deploy binaries | [`cmd/evobgp-ingest/main.go`](cmd/evobgp-ingest/main.go) | **low** (resource waste) | + +### Maintainability + +**Плюсы:** чёткое разделение слоёв (ARCH-01..10), `store.Backend`, OpenAPI как контракт, engineering rules, table-driven tests в birdfmt/pipeline. + +**Минусы:** +- Дублирование retry-логики (httpclient vs nodedispatch inline loop) +- Env-tuning разбросан (`EVOBGP_*` в 15+ местах без central config struct для pipeline) +- `Job` comment «персистенция в БД пока не подключена» устарел — hooks есть в [`bootstrap.go:67-92`](internal/httpapi/bootstrap.go) + +### Потенциальные баги и race conditions + +1. **`j.Meta` data race** — `-race` на `TestParallelModuleRefresh_*` + concurrent `GET /v1/jobs/{id}` polling +2. **Memory store alias** — `deploy.Run` читает `LastAppliedRevisionID` пока worker пишет +3. **peerLiveCache** возвращает slice без копии — [`peers_live.go:82-84`](internal/httpapi/peers_live.go) +4. **TOCTOU idempotency** — terminal job удаляется из `byIdempo`, повторный POST создаст новый job (by design, но клиент должен знать) + +### Error handling + +**Хорошо:** +- Префиксы ошибок (`httpclient:`, `birdfmt:`) +- HTTP 5xx через `writeProblem`, без raw `err.Error()` (ERR-01) +- `context.Context` в pipeline workers + +**Пробелы:** +- `runRollback` без `workContext` — не отменяется — [`worker.go:500+`](internal/jobs/worker.go) +- Prefetch/ingest: ошибки не логируются +- `mergeBirdPostApplyMeta` — `context.Background()` 8s, игнорирует job cancel + +--- + +## 3. Производительность + +### Блокирующие операции + +| Участок | Блокировка | Риск | +|---------|------------|------| +| `POST .../cdn-sources/preview` | Sync CDN fetch до 45s в HTTP handler | UI timeout, worker starvation | +| `GET /v1/peers/live` | N × agent HTTP, wg.Wait | Slow при многих speakers | +| Module refresh job | Sequential: ingest → render revision → optional deploy | Long job chain | +| `bird -p` / `birdc configure` | Subprocess в deploy | Disk I/O на ноде | + +### Неэффективные алгоритмы / лишние запросы + +- **Tenant refresh:** `aggregateTenantPrefixRowsAll` — parallel по модулям, но каждый модуль может refetch все CDN/ASN/DoH — [`aggregate.go:28+`](internal/pipeline/aggregate.go). Snapshot skip есть через `module_hash` — проверять hit rate в meta. +- **ASN resolve:** `PolitePause()` 150ms между AS — [`asnresolve/ripestat.go`](internal/asnresolve/ripestat.go) — при 50 AS = +7.5s minimum. +- **GetModulePrefixSnapshot** вызывается многократно в одном refresh (cdn_snapshot, collect_parallel) — potential duplicate DB reads. +- **Auth TouchAPIKeyLastUsed:** UPDATE на каждый request (async) — load на PG при high RPS. + +### Кэширование + +| Кэш | TTL | Gap | +|-----|-----|-----| +| ASN prefix cache | 1800s (`EVOBGP_ASN_CACHE_TTL_SEC`) | OK | +| CDN ETag in DB | Until 304/change | OK | +| Module prefix snapshot | Content-hash based skip | OK | +| peerLiveCache | In-memory, per-process | Не shared между API replicas; нет defensive copy | +| Circuit breaker state | Per-process | Не shared | + +### Конкретные улучшения + +```go +// 1. CDN preview — использовать upstreamHTTPDo вместо прямого Do +resp, err := pipeline.UpstreamHTTPDo(r.Context(), s.cdnHTTP, req) // extract upstreamHTTPDo + +// 2. Job.Meta — читать под lock или через Snapshot() +st := j.Snapshot() +mid, _ := st["meta"].(map[string]any)["module_id"].(string) + +// 3. Memory store — возвращать копии (как Postgres) +modCopy := *mod +return &modCopy, nil +``` + +--- + +## 4. Сетевое взаимодействие (критично) + +### Текущее состояние + +```mermaid +flowchart LR + subgraph resilient [Resilient path] + CDN[CDN fetch] + RIPE[RIPEstat] + CDN --> Breaker[Circuit breaker] + RIPE --> Breaker + Breaker --> Retry[DoWithRetry 3x linear 2s] + end + subgraph fragile [Fragile path] + DoH[DoH resolve] + Preview[CDN preview API] + AgentHealth[Agent health/bird] + DoH --> SingleDo[Single hc.Do] + Preview --> SingleDo + AgentHealth --> SingleDo + end + subgraph fallback [App-level fallback] + Stale[Stale snapshot] + SysDNS[System DNS] + DoH --> SysDNS + CDN --> Stale + RIPE --> Stale + end +``` + +| Upstream | Timeout | Retry | Breaker | Stale fallback | +|----------|---------|-------|---------|----------------| +| CDN ingest | 45s | 3× linear | per-host | yes | +| RIPEstat | 45s | 3× | per-host | yes + cache | +| DoH | 10s/profile | **no** | **no** | domain snapshot | +| CDN preview | 45s | **no** | **no** | N/A | +| Scheduler→API | 45s | 3× | no | N/A | +| Node dispatch | 30s | inline 3× | no | N/A | + +### Пробелы для блокировок провайдеров + +1. **DoH без retry** — transient timeout = fail; failover между profiles есть, но каждый profile — single shot +2. **429/408 не ретраятся** — только `>= 500` +3. **Нет jitter** — thundering herd при mass tenant refresh +4. **DNS rebinding TOCTOU** — SSRF check до fetch, HTTP dial без pinned IP — [`cdn_url.go:75-115`](internal/pipeline/cdn_url.go) +5. **Circuit breaker без half-open** — после 30s cooldown сразу full traffic — [`circuit.go:29-33`](internal/httpclient/circuit.go) +6. **Breaker per-process** — ingest container ≠ API container + +### Рекомендации для нестабильной сети + +| # | Изменение | Effort | Effect | +|---|-----------|--------|--------| +| N1 | DoH через `DoWithRetry` + optional breaker | Low | **High** для DOMAINS modules | +| N2 | Retry 429/503 с `Retry-After` + exponential backoff + jitter | Medium | **High** при rate limits | +| N3 | **Partial CDN failure** — continue с stale per-source, не fail whole module | Medium | **High** | +| N4 | Multiple DoH profiles + `failover` policy (already exists) — **документировать ops playbook** | Low | **High** (config, not code) | +| N5 | Pinned dialer / custom `Transport.DialContext` после SSRF resolve | Medium | **Medium** (SSRF hardening) | +| N6 | Proxy support (`HTTP_PROXY` / `EVOBGP_HTTP_PROXY`) для CDN/DoH | Medium | **High** в censored networks | +| N7 | Unify CDN preview на `upstreamHTTPDo` | Low | **Medium** | + +--- + +## 5. Устойчивость и надёжность + +### Graceful degradation + +**Работает:** +- `EVOBGP_STALE_ON_UPSTREAM_ERROR=1` — ASN/CDN/domain stale — [`collect_stale.go`](internal/pipeline/collect_stale.go) +- CDN 304 без local cache → forced full GET — [`cdn_snapshot.go:141-159`](internal/pipeline/cdn_snapshot.go) +- DoH → system DNS fallback — [`doh_resolve.go:75-93`](internal/pipeline/doh_resolve.go) +- Deploy: job succeed even if agent wake fails (drift detection) + +**Не работает / частично:** +- Один CDN source fail без cache → **весь module_refresh failed** — [`collect_parallel.go:221-223`](internal/pipeline/collect_parallel.go) +- Circuit open → immediate error, stale only if prior data exists +- API shutdown: HTTP drain 15s, **jobs не cancel/drain** — [`cmd/evobgp-api/main.go:67-72`](cmd/evobgp-api/main.go) + +### Сценарии отказов + +| Событие | Что произойдёт | +|---------|----------------| +| **Потеря CP↔PG** | Ready=false; running jobs fail; no new jobs persist audit reliably | +| **Потеря CP↔CDN** | Stale prefixes если были; иначе job fail; breaker opens 30s | +| **Потеря CP↔agent** | Deploy meta `dispatch_failed`; BIRD на старой ревизии; drift logs | +| **RIPEstat rate limit** | 429 → no retry → stale or fail | +| **Рост нагрузки** | Job queue; goroutine pile-up; PG pool exhaustion; `/metrics` shows queue depth | +| **API restart** | In-flight jobs lost; clients poll 404 or stale terminal state | + +--- + +## 6. Безопасность + +| ID | Finding | Severity | Fix | +|----|---------|----------|-----| +| S1 | `Bearer dev` → operator при demo-seed | **high** (misconfig) | `EVOBGP_SEED_DEMO=0` — [`auth.go:66-92`](internal/httpapi/auth.go) | +| S2 | API plain HTTP | **high** (ops) | TLS на edge (Traefik/nginx) | +| S3 | Ephemeral bundle key без `EVOBGP_BUNDLE_SEED_HEX` | **high** (ops) | Stable seed + pubkey на нодах | +| S4 | Compose defaults: weak PG password, `sslmode=disable` | **high** (ops) | Secrets manager, `sslmode=require` | +| S5 | `/metrics` без auth | **medium** | Network policy / mTLS | +| S6 | No rate limiting on auth | **medium** | Middleware limiter (e.g. per-IP) | +| S7 | CDN SSRF DNS rebinding | **medium** | Pinned dialer after resolve | +| S8 | `EVOBGP_CDN_ALLOW_PRIVATE=1` | **medium** | Never in prod | +| S9 | `EVOBGP_NODE_DISPATCH_INSECURE_TLS=1` | **medium** | Valid TLS to agent | +| S10 | Plaintext `EVOBGP_API_KEYS` in env | **medium** | DB keys via API | +| S11 | `editor` can cancel jobs | **low** | Restrict to operator | +| S12 | agent_secret `==` compare | **low** | `subtle.ConstantTimeCompare` | + +**SQL injection:** не обнаружено — параметризованные запросы в [`repository/`](internal/repository/). + +**Bundle crypto:** Ed25519 корректно; path traversal blocked в tar extract. + +--- + +## 7. Конкретные рекомендации (prioritized backlog) + +### High + +| # | Описание | Как исправить | +|---|----------|---------------| +| H1 | DoH без retry | Обернуть `hc.Do` в `DoWithRetry(ctx, hc, req, 3)` в [`refresh.go:288,360`](internal/pipeline/refresh.go) | +| H2 | Data race `Job.Meta` | Читать через `Snapshot()` или добавить `MetaLocked()` accessor | +| H3 | CDN source partial failure | В `collectCDNPrefixRows`: при err без stale — log warning + skip source вместо `return nil, r.err` (config flag `EVOBGP_CDN_PARTIAL_OK=1`) | +| H4 | Production checklist enforcement | CI/deploy validation: reject `SEED_DEMO=1`, require `BUNDLE_SEED_HEX` | +| H5 | Job queue HA roadmap | Persist queued jobs in PG + worker claim (`SELECT FOR UPDATE SKIP LOCKED`) или NATS — ARCH-04 | + +### Medium + +| # | Описание | Как исправить | +|---|----------|---------------| +| M1 | CDN preview bypass | [`routes_crud.go:267`](internal/httpapi/routes_crud.go) → `upstreamHTTPDo` | +| M2 | Retry 429/503 | Extend `DoWithRetry` status check + parse `Retry-After` | +| M3 | Graceful shutdown | On SIGTERM: `Registry.RequestCancelAll()` + wait workers with timeout | +| M4 | Auth goroutine storm | Worker pool или sync touch with debounce | +| M5 | HTTP proxy support | Custom Transport reading `EVOBGP_HTTP_PROXY` | +| M6 | Memory store copies | Defensive copy in Get/List (dev/test safety) | +| M7 | Rate limiting | `golang.org/x/time/rate` on auth middleware | + +### Low + +| # | Описание | Как исправить | +|---|----------|---------------| +| L1 | Jitter in backoff | `wait + rand.Intn(wait/2)` in DoWithRetry | +| L2 | Half-open breaker | Single probe request after cooldown | +| L3 | Remove dead `EVOBGP_DEV_INSECURE` from compose | Docs + compose cleanup | +| L4 | Prefetch error logging | `log.Printf` or structured log in prefetch | +| L5 | peerLiveCache defensive copy | `append([]T(nil), views...)` on store | + +--- + +## 8. Quick wins (максимальный эффект / минимум усилий) + +1. **Ops (0 code):** [`docs/production-checklist.md`](docs/production-checklist.md) — `SEED_DEMO=0`, `BUNDLE_SEED_HEX`, `DB_MAX_CONNS=25`, `JOB_MAX=16`, `COLLECT_CONCURRENCY=16`, TLS edge, restrict metrics +2. **DoH retry** — 5–10 строк в `refresh.go`, reuse existing `DoWithRetry` +3. **CDN preview → upstreamHTTPDo** — 1 line change in handler +4. **Job.Meta read fix** — replace 4 reads in `worker.go` with `Snapshot()` parsing +5. **Log prefetch failures** — visibility без изменения behavior +6. **Document DoH failover playbook** — multiple profiles (Cloudflare, Google, Quad9) + `failover` policy for censored regions +7. **Run `go test -race ./internal/jobs/...`** in CI — catch Meta race +8. **Prefer `evobgp-all`** over split reference for <20 tenants — eliminates Registry split bug + +--- + +## Диаграмма: refresh под сетевым stress + +```mermaid +sequenceDiagram + participant Op as Operator + participant API as evobgp_api + participant Job as module_refresh + participant CDN as CDN_upstream + participant PG as PostgreSQL + + Op->>API: POST /modules/id/refresh + API->>Job: Enqueue + Job->>CDN: GET with ETag + alt CDN timeout or 5xx + CDN-->>Job: error after 3 retries + Job->>PG: load prior snapshot + alt stale exists + Job->>PG: CreateRenderRevision stale + Job-->>API: succeeded degraded + else no stale + Job-->>API: failed + end + else CDN 200 + CDN-->>Job: new prefixes + Job->>PG: CreateRenderRevision + end +``` + +--- + +## Итоговая оценка зрелости + +| Область | Оценка | Комментарий | +|---------|--------|-------------| +| Архитектура | 7/10 | Чистые слои; HA/API scaling — слабое место | +| Сеть/resilience | 6/10 | CDN/ASN хорошо; DoH/preview — пробелы | +| Concurrency | 6/10 | Registry продуман; Meta race, shutdown | +| Performance | 7/10 | Parallel collect, caching; tuning needed at scale | +| Security | 6/10 | Crypto OK; ops/config risks dominate | +| Maintainability | 8/10 | Docs, rules, OpenAPI, tests | + +**Вердикт:** проект **готов для 10+ клиентов в single-CP deployment** (`evobgp-all` + PostgreSQL + production checklist) при условии ops discipline. Для **multi-CP HA** и **агрессивных сетевых блокировок** — приоритет: DoH retry, partial CDN failure, distributed job queue, HTTP proxy. diff --git a/cmd/evobgp-all/main.go b/cmd/evobgp-all/main.go index 2046f42..b8bcd11 100644 --- a/cmd/evobgp-all/main.go +++ b/cmd/evobgp-all/main.go @@ -12,6 +12,7 @@ import ( "evobgp/internal/birdfmt" "evobgp/internal/config" + "evobgp/internal/dbcli" "evobgp/internal/deploy" "evobgp/internal/httpapi" "evobgp/internal/ingest" @@ -24,6 +25,9 @@ import ( // microVPS entrypoint: один процесс — HTTP API и фоновые воркеры scheduler, ingest, render, deploy (общий store и jobs.Registry). func main() { + if len(os.Args) > 1 && os.Args[1] == "db" { + os.Exit(dbcli.Run(os.Args[2:])) + } cfg := config.Load() opts := httpapi.Options{ APIKeys: os.Getenv("EVOBGP_API_KEYS"), @@ -52,6 +56,7 @@ func main() { go render.Run(ctx, renderDeps) go deploy.Run(ctx, deployDeps) + srv.StartBackground(ctx) startBirdMetricsPoller(ctx) httpSrv := &http.Server{ diff --git a/cmd/evobgp-api/main.go b/cmd/evobgp-api/main.go index 91f2683..357df9f 100644 --- a/cmd/evobgp-api/main.go +++ b/cmd/evobgp-api/main.go @@ -12,6 +12,7 @@ import ( "evobgp/internal/birdfmt" "evobgp/internal/config" + "evobgp/internal/dbcli" "evobgp/internal/httpapi" "evobgp/internal/observability" "evobgp/internal/platform" @@ -19,6 +20,9 @@ import ( ) func main() { + if len(os.Args) > 1 && os.Args[1] == "db" { + os.Exit(dbcli.Run(os.Args[2:])) + } cfg := config.Load() seedDemo := os.Getenv("EVOBGP_SEED_DEMO") != "0" opts := httpapi.Options{ @@ -39,6 +43,7 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + srv.StartBackground(ctx) startBirdMetricsPoller(ctx) httpSrv := &http.Server{ diff --git a/docs/access.md b/docs/access.md index 211cab2..155a48b 100644 --- a/docs/access.md +++ b/docs/access.md @@ -52,6 +52,18 @@ opkey|01ARZ3NDEKTSV4RRFFQ69G5FAV|operator,nodekey|01ARZ3NDEKTSV4RRFFQ69G5FAV|nod **Запрещено** в продакшене: не оставляйте demo-seed с известным токеном `dev` на боевых данных. Переменная `EVOBGP_DEV_INSECURE` в текущей версии **не влияет** на аутентификацию (оставлена в compose для совместимости; не включайте в production — см. SEC-02 в инженерных правилах). +### PostgreSQL monitoring и maintenance (control plane) + +При `EVOBGP_DATABASE_URL` (не memory backend): + +| Операция | Минимальная роль | +|----------|------------------| +| `GET /v1/monitoring/postgres/*`, `GET /v1/monitoring/correlation` | viewer | +| `POST /v1/postgres/vacuum`, `vacuum-analyze`, `analyze`, `reindex`, `cleanup` | **operator** (async job, rate limit 60s на kind) | +| `GET /v1/postgres/maintenance/logs` | viewer | + +Метрики **instance-level** (не per-tenant). CLI: `evobgp-api db …` / `evobgp-all db …`. + ### Синхронные «тяжёлые» GET (control plane) - `POST /v1/modules/{module_id}/cdn-sources/preview` — загрузка CDN в том же HTTP-запросе (лимит тела ~8 MiB, см. OpenAPI). diff --git a/docs/db-diagnostics.md b/docs/db-diagnostics.md index 3f1aeb9..263fcb5 100644 --- a/docs/db-diagnostics.md +++ b/docs/db-diagnostics.md @@ -8,6 +8,21 @@ Runbook для оценки объёма БД и узких мест **пере psql "$EVOBGP_DATABASE_URL" ``` +## HTTP API (панель / мониторинг) + +При подключённом PostgreSQL control plane отдаёт instance-level метрики (роль **viewer+**): + +- `GET /v1/monitoring/postgres/overview` — подключения, TPS, cache hit, размер БД +- `GET /v1/monitoring/postgres/queries` — top queries (`pg_stat_statements`, если extension включён) +- `GET /v1/monitoring/postgres/locks`, `/tables`, `/recommendations` +- `GET /v1/monitoring/correlation?window=60` — корреляция refresh jobs и cache hit + +Обслуживание (**operator**, async `202` + `job_id`): `POST /v1/postgres/vacuum`, `vacuum-analyze`, `analyze`, `reindex`, `cleanup`; журнал `GET /v1/postgres/maintenance/logs`. + +CLI на CP: `evobgp-api db report|vacuum|analyze|cleanup` (см. `internal/dbcli`). + +Миграция `000023` создаёт `pg_stat_statements`; в production может потребоваться `shared_preload_libraries` и перезапуск Postgres. + ## 1. Размеры таблиц и индексов ```sql diff --git a/docs/openapi.yaml b/docs/openapi.yaml index cf9fc8b..73641c1 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -49,6 +49,8 @@ tags: description: Управление API-ключами tenant (operator). Секрет возвращается только при создании и ротации. - name: Auth description: Сессия текущего API-ключа (tenant и роль). + - name: Monitoring + description: Наблюдаемость PostgreSQL и корреляция (instance-level, viewer+). Maintenance — operator. security: - bearerAuth: [] @@ -910,6 +912,64 @@ components: type: string additionalProperties: true + PostgresOverview: + type: object + description: Instance-level PostgreSQL snapshot (GET /v1/monitoring/postgres/overview). + additionalProperties: true + + PostgresQueriesResponse: + type: object + properties: + collected_at: + type: string + format: date-time + source: + type: string + enum: [live, snapshot] + items: + type: array + items: + type: object + additionalProperties: true + + PostgresRecommendations: + type: object + properties: + collected_at: + type: string + format: date-time + items: + type: array + items: + type: object + properties: + severity: + type: string + code: + type: string + title: + type: string + detail: + type: string + refs: + type: array + items: + type: string + + PostgresMaintenanceBody: + type: object + properties: + table: + type: string + dry_run: + type: boolean + default: false + policy: + type: string + description: job_audit_retention | asn_cache_retention + limit: + type: integer + BirdLocalStatus: type: object description: Статус локального BIRD на хосте API (GET /v1/bird/status). @@ -3135,6 +3195,278 @@ paths: default: $ref: "#/components/responses/DefaultProblem" + /v1/monitoring/postgres/overview: + get: + tags: [Monitoring] + summary: PostgreSQL overview (instance-level) + operationId: getPostgresOverview + parameters: + - $ref: "#/components/parameters/TenantId" + responses: + "200": + description: Успешно. + content: + application/json: + schema: + $ref: "#/components/schemas/PostgresOverview" + "503": + description: PostgreSQL backend не подключён. + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/monitoring/postgres/queries: + get: + tags: [Monitoring] + summary: Top queries (pg_stat_statements or snapshot) + operationId: getPostgresQueries + parameters: + - $ref: "#/components/parameters/TenantId" + - $ref: "#/components/parameters/Limit" + responses: + "200": + description: Успешно. + content: + application/json: + schema: + $ref: "#/components/schemas/PostgresQueriesResponse" + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/monitoring/postgres/locks: + get: + tags: [Monitoring] + summary: Active locks + operationId: getPostgresLocks + parameters: + - $ref: "#/components/parameters/TenantId" + responses: + "200": + description: Успешно. + content: + application/json: + schema: + type: object + properties: + items: + type: array + items: + type: object + additionalProperties: true + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/monitoring/postgres/tables: + get: + tags: [Monitoring] + summary: Table sizes and scan stats + operationId: getPostgresTables + parameters: + - $ref: "#/components/parameters/TenantId" + - $ref: "#/components/parameters/Limit" + responses: + "200": + description: Успешно. + content: + application/json: + schema: + type: object + properties: + items: + type: array + items: + type: object + additionalProperties: true + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/monitoring/postgres/recommendations: + get: + tags: [Monitoring] + summary: Heuristic optimization recommendations + operationId: getPostgresRecommendations + parameters: + - $ref: "#/components/parameters/TenantId" + responses: + "200": + description: Успешно. + content: + application/json: + schema: + $ref: "#/components/schemas/PostgresRecommendations" + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/monitoring/correlation: + get: + tags: [Monitoring] + summary: Timeline correlation (jobs vs cache hit) + operationId: getMonitoringCorrelation + parameters: + - $ref: "#/components/parameters/TenantId" + - name: window + in: query + schema: + type: integer + default: 60 + description: Window in minutes (max 1440). + responses: + "200": + description: Успешно. + content: + application/json: + schema: + type: object + additionalProperties: true + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/postgres/vacuum: + post: + tags: [Monitoring] + summary: VACUUM (async job, operator) + operationId: postPostgresVacuum + parameters: + - $ref: "#/components/parameters/TenantId" + - $ref: "#/components/parameters/IdempotencyKey" + requestBody: + content: + application/json: + schema: + $ref: "#/components/schemas/PostgresMaintenanceBody" + responses: + "202": + description: Задача поставлена. + content: + application/json: + schema: + $ref: "#/components/schemas/AsyncJobAccepted" + "403": + $ref: "#/components/responses/Forbidden" + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/postgres/vacuum-analyze: + post: + tags: [Monitoring] + summary: VACUUM ANALYZE (async job, operator) + operationId: postPostgresVacuumAnalyze + parameters: + - $ref: "#/components/parameters/TenantId" + - $ref: "#/components/parameters/IdempotencyKey" + requestBody: + content: + application/json: + schema: + $ref: "#/components/schemas/PostgresMaintenanceBody" + responses: + "202": + description: Задача поставлена. + content: + application/json: + schema: + $ref: "#/components/schemas/AsyncJobAccepted" + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/postgres/analyze: + post: + tags: [Monitoring] + summary: ANALYZE (async job, operator) + operationId: postPostgresAnalyze + parameters: + - $ref: "#/components/parameters/TenantId" + - $ref: "#/components/parameters/IdempotencyKey" + requestBody: + content: + application/json: + schema: + $ref: "#/components/schemas/PostgresMaintenanceBody" + responses: + "202": + description: Задача поставлена. + content: + application/json: + schema: + $ref: "#/components/schemas/AsyncJobAccepted" + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/postgres/reindex: + post: + tags: [Monitoring] + summary: REINDEX TABLE (async job, operator) + operationId: postPostgresReindex + parameters: + - $ref: "#/components/parameters/TenantId" + - $ref: "#/components/parameters/IdempotencyKey" + requestBody: + content: + application/json: + schema: + $ref: "#/components/schemas/PostgresMaintenanceBody" + responses: + "202": + description: Задача поставлена. + content: + application/json: + schema: + $ref: "#/components/schemas/AsyncJobAccepted" + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/postgres/cleanup: + post: + tags: [Monitoring] + summary: Retention cleanup (async job, operator) + operationId: postPostgresCleanup + parameters: + - $ref: "#/components/parameters/TenantId" + - $ref: "#/components/parameters/IdempotencyKey" + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/PostgresMaintenanceBody" + responses: + "202": + description: Задача поставлена. + content: + application/json: + schema: + $ref: "#/components/schemas/AsyncJobAccepted" + default: + $ref: "#/components/responses/DefaultProblem" + + /v1/postgres/maintenance/logs: + get: + tags: [Monitoring] + summary: Maintenance audit log + operationId: listPostgresMaintenanceLogs + parameters: + - $ref: "#/components/parameters/TenantId" + - $ref: "#/components/parameters/Cursor" + - $ref: "#/components/parameters/Limit" + responses: + "200": + description: Успешно. + content: + application/json: + schema: + type: object + properties: + items: + type: array + items: + type: object + additionalProperties: true + next_cursor: + type: string + has_more: + type: boolean + default: + $ref: "#/components/responses/DefaultProblem" + /v1/settings: get: tags: [Settings] diff --git a/internal/dbcli/dbcli.go b/internal/dbcli/dbcli.go new file mode 100644 index 0000000..bac8b23 --- /dev/null +++ b/internal/dbcli/dbcli.go @@ -0,0 +1,209 @@ +// Package dbcli implements control-plane PostgreSQL maintenance CLI (HTTP or local DSN). +package dbcli + +import ( + "bytes" + "context" + "encoding/json" + "flag" + "fmt" + "io" + "net/http" + "os" + "strings" + "time" + + "evobgp/internal/db" + "evobgp/internal/httpclient" + "evobgp/internal/pgmonitor" +) + +// Run executes db subcommands; args exclude program name and "db". +func Run(args []string) int { + if len(args) == 0 { + printUsage() + return 2 + } + switch args[0] { + case "report": + return cmdReport(args[1:]) + case "vacuum": + return cmdMaint(args[1:], "vacuum", "/v1/postgres/vacuum") + case "analyze": + return cmdMaint(args[1:], "analyze", "/v1/postgres/analyze") + case "cleanup": + return cmdCleanup(args[1:]) + default: + fmt.Fprintf(os.Stderr, "dbcli: unknown command %q\n", args[0]) + printUsage() + return 2 + } +} + +func printUsage() { + fmt.Fprintln(os.Stderr, `usage: + evobgp-api db report [--api-url URL] [--token TOKEN] [--format json] + evobgp-api db vacuum [--table NAME] [--dry-run] [--api-url URL] [--token TOKEN] + evobgp-api db analyze [--table NAME] [--dry-run] [--api-url URL] [--token TOKEN] + evobgp-api db cleanup --policy NAME [--dry-run] [--limit N] [--api-url URL] [--token TOKEN] +Local break-glass: set EVOBGP_DATABASE_URL (report only uses direct SQL).`) +} + +func cmdReport(args []string) int { + fs := flag.NewFlagSet("report", flag.ExitOnError) + apiURL := fs.String("api-url", "", "control plane base URL") + token := fs.String("token", "", "Bearer token (operator)") + format := fs.String("format", "json", "output format (json)") + _ = fs.Parse(args) + + if dsn := strings.TrimSpace(os.Getenv("EVOBGP_DATABASE_URL")); dsn != "" && *apiURL == "" { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + pool, err := db.OpenPostgresPool(ctx, dsn) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + defer pool.Close() + svc := pgmonitor.NewService(pool) + ov, err := svc.Overview(ctx) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + return writeJSONStdout(ov, *format) + } + if *apiURL == "" || *token == "" { + fmt.Fprintln(os.Stderr, "report: --api-url and --token required without EVOBGP_DATABASE_URL") + return 2 + } + body, err := apiGET(*apiURL, *token, "/v1/monitoring/postgres/overview") + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + var pretty any + if err := json.Unmarshal(body, &pretty); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + return writeJSONStdout(pretty, *format) +} + +func cmdMaint(args []string, _ string, path string) int { + fs := flag.NewFlagSet("maint", flag.ExitOnError) + table := fs.String("table", "", "table name") + dryRun := fs.Bool("dry-run", false, "dry run only") + apiURL := fs.String("api-url", "", "control plane base URL") + token := fs.String("token", "", "Bearer token (operator)") + _ = fs.Parse(args) + if *apiURL == "" || *token == "" { + fmt.Fprintln(os.Stderr, "maintenance: --api-url and --token are required") + return 2 + } + payload := map[string]any{"dry_run": *dryRun} + if *table != "" { + payload["table"] = *table + } + body, err := apiPOST(*apiURL, *token, path, payload) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + return writeRawJSON(body) +} + +func cmdCleanup(args []string) int { + fs := flag.NewFlagSet("cleanup", flag.ExitOnError) + policy := fs.String("policy", "", "cleanup policy name") + dryRun := fs.Bool("dry-run", true, "dry run") + limit := fs.Int("limit", 10000, "max rows") + apiURL := fs.String("api-url", "", "control plane base URL") + token := fs.String("token", "", "Bearer token (operator)") + _ = fs.Parse(args) + if *policy == "" { + fmt.Fprintln(os.Stderr, "cleanup: --policy is required") + return 2 + } + if *apiURL == "" || *token == "" { + fmt.Fprintln(os.Stderr, "cleanup: --api-url and --token are required") + return 2 + } + payload := map[string]any{"policy": *policy, "dry_run": *dryRun, "limit": *limit} + body, err := apiPOST(*apiURL, *token, "/v1/postgres/cleanup", payload) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + return writeRawJSON(body) +} + +func apiGET(base, token, path string) ([]byte, error) { + u := strings.TrimRight(base, "/") + path + req, err := http.NewRequest(http.MethodGet, u, nil) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+token) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + resp, err := httpclient.DoWithRetry(ctx, httpclient.New(60*time.Second), req, 3) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + b, _ := io.ReadAll(resp.Body) + if resp.StatusCode >= 300 { + return nil, fmt.Errorf("dbcli: GET %s: %s: %s", path, resp.Status, strings.TrimSpace(string(b))) + } + return b, nil +} + +func apiPOST(base, token, path string, payload map[string]any) ([]byte, error) { + b, err := json.Marshal(payload) + if err != nil { + return nil, err + } + u := strings.TrimRight(base, "/") + path + req, err := http.NewRequest(http.MethodPost, u, bytes.NewReader(b)) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + resp, err := httpclient.DoWithRetry(ctx, httpclient.New(60*time.Second), req, 3) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + out, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusAccepted && resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("dbcli: POST %s: %s: %s", path, resp.Status, strings.TrimSpace(string(out))) + } + return out, nil +} + +func writeJSONStdout(v any, format string) int { + if format != "json" { + fmt.Fprintln(os.Stderr, "only json format supported") + return 2 + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + if err := enc.Encode(v); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + return 0 +} + +func writeRawJSON(b []byte) int { + var v any + if err := json.Unmarshal(b, &v); err != nil { + _, _ = os.Stdout.Write(b) + return 0 + } + return writeJSONStdout(v, "json") +} diff --git a/internal/httpapi/bootstrap.go b/internal/httpapi/bootstrap.go index 28b02b3..8e3a5c7 100644 --- a/internal/httpapi/bootstrap.go +++ b/internal/httpapi/bootstrap.go @@ -51,7 +51,7 @@ func BootstrapWorkers(ctx context.Context, opts Options) (store.Backend, *jobs.R } cdnHTTP := NewCDNHTTPClient() - wk := &jobs.Worker{Store: backend, HTTPClient: cdnHTTP} + wk := &jobs.Worker{Store: backend, PgPool: pool, HTTPClient: cdnHTTP} reg := jobs.NewRegistry(wk.Process) wk.Registry = reg if pool != nil { diff --git a/internal/httpapi/routes.go b/internal/httpapi/routes.go index 6e94479..2641718 100644 --- a/internal/httpapi/routes.go +++ b/internal/httpapi/routes.go @@ -76,6 +76,8 @@ func (s *Server) registerV1(m *http.ServeMux) { m.HandleFunc("GET /speakers/{speaker_id}/bundle/{revision_id}", s.handleNodeBundle) m.HandleFunc("POST /nodes/enroll", s.handleNodeEnroll) s.registerCRUDRoutes(m) + s.registerPostgresMonitoringRoutes(m) + s.registerPostgresMaintenanceRoutes(m) } func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { diff --git a/internal/httpapi/routes_postgres_maintenance.go b/internal/httpapi/routes_postgres_maintenance.go new file mode 100644 index 0000000..797778a --- /dev/null +++ b/internal/httpapi/routes_postgres_maintenance.go @@ -0,0 +1,203 @@ +package httpapi + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "sync" + "time" + + "evobgp/internal/jobs" + "evobgp/internal/pgmonitor" +) + +var ( + pgMaintRateMu sync.Mutex + pgMaintLastByTK = map[string]time.Time{} +) + +func (s *Server) registerPostgresMaintenanceRoutes(m *http.ServeMux) { + m.HandleFunc("POST /postgres/vacuum", s.handlePostgresVacuum) + m.HandleFunc("POST /postgres/vacuum-analyze", s.handlePostgresVacuumAnalyze) + m.HandleFunc("POST /postgres/analyze", s.handlePostgresAnalyze) + m.HandleFunc("POST /postgres/reindex", s.handlePostgresReindex) + m.HandleFunc("POST /postgres/cleanup", s.handlePostgresCleanup) + m.HandleFunc("GET /postgres/maintenance/logs", s.handlePostgresMaintenanceLogs) +} + +func (s *Server) requireOperatorStrict(w http.ResponseWriter, a Auth) bool { + if strings.ToLower(a.Role) != "operator" { + writeProblem(w, http.StatusForbidden, "Forbidden", "operator role required") + return false + } + return true +} + +func (s *Server) checkPgMaintRateLimit(tenantID, kind string) bool { + key := tenantID + ":" + kind + pgMaintRateMu.Lock() + defer pgMaintRateMu.Unlock() + if t, ok := pgMaintLastByTK[key]; ok && time.Since(t) < 60*time.Second { + return false + } + pgMaintLastByTK[key] = time.Now().UTC() + return true +} + +type pgMaintBody struct { + Table string `json:"table"` + DryRun bool `json:"dry_run"` + Index string `json:"index"` + Policy string `json:"policy"` + Limit int `json:"limit"` +} + +func (s *Server) decodePgMaintBody(r *http.Request) (pgMaintBody, bool) { + var body pgMaintBody + if r.Body == nil || r.ContentLength == 0 { + return body, true + } + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&body); err != nil && err != io.EOF { + return body, false + } + return body, true +} + +func (s *Server) enqueuePostgresMaint(w http.ResponseWriter, r *http.Request, a Auth, kind string, meta map[string]any) { + if !s.requirePostgres(w) || !s.requireOperatorStrict(w, a) { + return + } + if !s.checkPgMaintRateLimit(a.TenantID, kind) { + writeProblem(w, http.StatusTooManyRequests, "Too Many Requests", "wait before repeating this maintenance operation") + return + } + idem := strings.TrimSpace(r.Header.Get("Idempotency-Key")) + var idemPtr *string + if idem != "" { + idemPtr = &idem + } + meta["actor_prefix"] = actorPrefix(a) + j, _, err := s.jobs.Enqueue(a.TenantID, kind, idemPtr, nil, meta) + if err != nil { + writeInternalError(w, "postgres_maint_enqueue", err) + return + } + w.Header().Set("Location", "/v1/jobs/"+j.ID) + snap := j.Snapshot() + writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]}) +} + +func (s *Server) handlePostgresVacuum(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok { + writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") + return + } + body, ok2 := s.decodePgMaintBody(r) + if !ok2 { + writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) + return + } + s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresVacuum, map[string]any{ + "table": body.Table, "dry_run": body.DryRun, "job_title": "PostgreSQL VACUUM", + }) +} + +func (s *Server) handlePostgresVacuumAnalyze(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok { + writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") + return + } + body, ok2 := s.decodePgMaintBody(r) + if !ok2 { + writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) + return + } + s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresVacuumAnalyze, map[string]any{ + "table": body.Table, "dry_run": body.DryRun, "job_title": "PostgreSQL VACUUM ANALYZE", + }) +} + +func (s *Server) handlePostgresAnalyze(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok { + writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") + return + } + body, ok2 := s.decodePgMaintBody(r) + if !ok2 { + writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) + return + } + s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresAnalyze, map[string]any{ + "table": body.Table, "dry_run": body.DryRun, "job_title": "PostgreSQL ANALYZE", + }) +} + +func (s *Server) handlePostgresReindex(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok { + writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") + return + } + body, ok2 := s.decodePgMaintBody(r) + if !ok2 { + writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) + return + } + table := body.Table + if table == "" { + table = body.Index + } + s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresReindex, map[string]any{ + "table": table, "dry_run": body.DryRun, "job_title": "PostgreSQL REINDEX", + }) +} + +func (s *Server) handlePostgresCleanup(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok { + writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") + return + } + body, ok2 := s.decodePgMaintBody(r) + if !ok2 { + writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) + return + } + if strings.TrimSpace(body.Policy) == "" { + writeProblem(w, http.StatusBadRequest, "Bad Request", "policy is required") + return + } + s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresCleanup, map[string]any{ + "policy": body.Policy, "dry_run": body.DryRun, "limit": body.Limit, + "job_title": "PostgreSQL cleanup", + }) +} + +func (s *Server) handlePostgresMaintenanceLogs(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) { + return + } + cursor := r.URL.Query().Get("cursor") + limit := parseLimitQuery(r, 20, 100) + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + items, next, hasMore, err := pgmonitor.ListMaintenanceLogs(ctx, s.pgMonitor.Pool(), cursor, limit) + if err != nil { + writeInternalError(w, "postgres_maint_logs", err) + return + } + writeJSON(w, http.StatusOK, map[string]any{"items": items, "next_cursor": next, "has_more": hasMore}) +} + +func actorPrefix(a Auth) string { + if len(a.Token) >= 8 { + return a.Token[:8] + } + return a.Role +} diff --git a/internal/httpapi/routes_postgres_monitoring.go b/internal/httpapi/routes_postgres_monitoring.go new file mode 100644 index 0000000..522e8d3 --- /dev/null +++ b/internal/httpapi/routes_postgres_monitoring.go @@ -0,0 +1,130 @@ +package httpapi + +import ( + "context" + "net/http" + "strconv" + "time" +) + +func (s *Server) registerPostgresMonitoringRoutes(m *http.ServeMux) { + m.HandleFunc("GET /monitoring/postgres/overview", s.handlePostgresOverview) + m.HandleFunc("GET /monitoring/postgres/queries", s.handlePostgresQueries) + m.HandleFunc("GET /monitoring/postgres/locks", s.handlePostgresLocks) + m.HandleFunc("GET /monitoring/postgres/tables", s.handlePostgresTables) + m.HandleFunc("GET /monitoring/postgres/recommendations", s.handlePostgresRecommendations) + m.HandleFunc("GET /monitoring/correlation", s.handleMonitoringCorrelation) +} + +func (s *Server) requirePostgres(w http.ResponseWriter) bool { + if s.pgMonitor == nil { + writeProblem(w, http.StatusServiceUnavailable, "Unavailable", "postgresql backend required") + return false + } + return true +} + +func parseLimitQuery(r *http.Request, def, max int) int { + if v := r.URL.Query().Get("limit"); v != "" { + if n, err := strconv.Atoi(v); err == nil { + return n + } + } + return def +} + +func (s *Server) handlePostgresOverview(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second) + defer cancel() + out, err := s.pgMonitor.Overview(ctx) + if err != nil { + writeInternalError(w, "postgres_overview", err) + return + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) handlePostgresQueries(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second) + defer cancel() + out, err := s.pgMonitor.TopQueries(ctx, parseLimitQuery(r, 20, 100)) + if err != nil { + writeInternalError(w, "postgres_queries", err) + return + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) handlePostgresLocks(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second) + defer cancel() + out, err := s.pgMonitor.Locks(ctx) + if err != nil { + writeInternalError(w, "postgres_locks", err) + return + } + writeJSON(w, http.StatusOK, map[string]any{"items": out}) +} + +func (s *Server) handlePostgresTables(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second) + defer cancel() + out, err := s.pgMonitor.Tables(ctx, parseLimitQuery(r, 20, 100)) + if err != nil { + writeInternalError(w, "postgres_tables", err) + return + } + writeJSON(w, http.StatusOK, map[string]any{"items": out}) +} + +func (s *Server) handlePostgresRecommendations(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + out, err := s.pgMonitor.Recommendations(ctx) + if err != nil { + writeInternalError(w, "postgres_recommendations", err) + return + } + writeJSON(w, http.StatusOK, out) +} + +func (s *Server) handleMonitoringCorrelation(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok || !s.requireAtLeast(w, a, "viewer") || !s.requirePostgres(w) { + return + } + window := 60 + if v := r.URL.Query().Get("window"); v != "" { + if n, err := strconv.Atoi(v); err == nil { + window = n + } + } + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + out, err := s.pgMonitor.Correlation(ctx, window) + if err != nil { + writeInternalError(w, "monitoring_correlation", err) + return + } + writeJSON(w, http.StatusOK, out) +} diff --git a/internal/httpapi/routes_postgres_test.go b/internal/httpapi/routes_postgres_test.go new file mode 100644 index 0000000..ad7346b --- /dev/null +++ b/internal/httpapi/routes_postgres_test.go @@ -0,0 +1,21 @@ +package httpapi + +import ( + "net/http" + "net/http/httptest" + "testing" +) + +func TestPostgresOverviewMemoryBackend503(t *testing.T) { + srv, err := New(Options{SeedDemo: true}) + if err != nil { + t.Fatal(err) + } + req := httptest.NewRequest(http.MethodGet, "/v1/monitoring/postgres/overview", nil) + req.Header.Set("Authorization", "Bearer dev") + rec := httptest.NewRecorder() + srv.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String()) + } +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 6a80799..f163150 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -10,6 +10,7 @@ import ( "strings" "evobgp/internal/jobs" + "evobgp/internal/pgmonitor" "evobgp/internal/store" "github.com/jackc/pgx/v5/pgxpool" @@ -19,6 +20,7 @@ import ( type Server struct { store store.Backend pgPool *pgxpool.Pool + pgMonitor *pgmonitor.Service jobs *jobs.Registry bundlePriv ed25519.PrivateKey keyResolver *apiKeyResolver @@ -63,9 +65,14 @@ func New(opts Options) (*Server, error) { if err != nil { return nil, err } + var pgMon *pgmonitor.Service + if pool != nil { + pgMon = pgmonitor.NewService(pool) + } s := &Server{ store: backend, pgPool: pool, + pgMonitor: pgMon, jobs: reg, bundlePriv: priv, keyResolver: resolver, @@ -89,3 +96,10 @@ func (s *Server) Store() store.Backend { return s.store } // Jobs exposes the in-process async job registry (for scheduler / evobgp-all). func (s *Server) Jobs() *jobs.Registry { return s.jobs } + +// StartBackground starts PostgreSQL monitoring scheduler until ctx is cancelled. +func (s *Server) StartBackground(ctx context.Context) { + if s != nil && s.pgPool != nil { + pgmonitor.StartScheduler(ctx, s.pgPool) + } +} diff --git a/internal/jobs/postgres_worker.go b/internal/jobs/postgres_worker.go new file mode 100644 index 0000000..0183742 --- /dev/null +++ b/internal/jobs/postgres_worker.go @@ -0,0 +1,169 @@ +package jobs + +import ( + "fmt" + "strings" + + "evobgp/internal/pgmonitor" +) + +func (w *Worker) pgService() *pgmonitor.Service { + if w == nil || w.PgPool == nil { + return nil + } + return pgmonitor.NewService(w.PgPool) +} + +func (w *Worker) runPostgresMetricsRefresh(j *Job) { + s := w.pgService() + if s == nil { + j.Fail("postgresql not configured") + return + } + ctx, cancel := j.workContext() + defer cancel() + if err := s.RefreshMetricsSnapshot(ctx); err != nil { + j.Fail(err.Error()) + return + } + j.Succeed() +} + +func (w *Worker) runPostgresSlowQueryAgg(j *Job) { + s := w.pgService() + if s == nil { + j.Fail("postgresql not configured") + return + } + ctx, cancel := j.workContext() + defer cancel() + if err := s.AggregateSlowQueries(ctx, 30); err != nil { + j.Fail(err.Error()) + return + } + j.Succeed() +} + +func (w *Worker) runPostgresTableBloat(j *Job) { + s := w.pgService() + if s == nil { + j.Fail("postgresql not configured") + return + } + ctx, cancel := j.workContext() + defer cancel() + if err := s.EstimateTableBloat(ctx); err != nil { + j.Fail(err.Error()) + return + } + j.Succeed() +} + +func (w *Worker) runPostgresIndexUsage(j *Job) { + s := w.pgService() + if s == nil { + j.Fail("postgresql not configured") + return + } + ctx, cancel := j.workContext() + defer cancel() + if err := s.AnalyzeIndexUsage(ctx); err != nil { + j.Fail(err.Error()) + return + } + j.Succeed() +} + +func (w *Worker) runPostgresAutovacuumLag(j *Job) { + s := w.pgService() + if s == nil { + j.Fail("postgresql not configured") + return + } + ctx, cancel := j.workContext() + defer cancel() + if err := s.DetectAutovacuumLag(ctx); err != nil { + j.Fail(err.Error()) + return + } + j.Succeed() +} + +func (w *Worker) runPostgresMaint(j *Job, kind string) { + if w == nil || w.PgPool == nil { + j.Fail("postgresql not configured") + return + } + table, _ := j.Meta["table"].(string) + dryRun, _ := j.Meta["dry_run"].(bool) + actor, _ := j.Meta["actor_prefix"].(string) + ctx, cancel := j.workContext() + defer cancel() + auditID, _ := pgmonitor.InsertMaintenanceAudit(ctx, w.PgPool, j.TenantID, actor, kind, table, dryRun) + detail, err := pgmonitor.ExecMaintenance(ctx, w.PgPool, kind, table, dryRun) + var errMsg *string + status := StatusSucceeded + if err != nil { + s := err.Error() + errMsg = &s + status = StatusFailed + j.Fail(s) + } else { + j.mergeMeta(map[string]any{"maintenance": detail, "audit_id": auditID}) + j.Succeed() + } + if auditID != "" { + _ = pgmonitor.FinishMaintenanceAudit(ctx, w.PgPool, auditID, status, detail, errMsg) + } +} + +func (w *Worker) runPostgresCleanup(j *Job) { + if w == nil || w.PgPool == nil { + j.Fail("postgresql not configured") + return + } + policy, _ := j.Meta["policy"].(string) + dryRun, _ := j.Meta["dry_run"].(bool) + limit := 0 + if v, ok := j.Meta["limit"].(float64); ok { + limit = int(v) + } + actor, _ := j.Meta["actor_prefix"].(string) + ctx, cancel := j.workContext() + defer cancel() + auditID, _ := pgmonitor.InsertMaintenanceAudit(ctx, w.PgPool, j.TenantID, actor, "cleanup", policy, dryRun) + detail, err := pgmonitor.RunCleanup(ctx, w.PgPool, strings.TrimSpace(policy), dryRun, limit) + var errMsg *string + status := StatusSucceeded + if err != nil { + s := err.Error() + errMsg = &s + status = StatusFailed + j.Fail(s) + } else { + j.mergeMeta(map[string]any{"cleanup": detail, "audit_id": auditID}) + j.Succeed() + } + if auditID != "" { + _ = pgmonitor.FinishMaintenanceAudit(ctx, w.PgPool, auditID, status, detail, errMsg) + } +} + +// EnqueuePostgresAnalyzerJobs enqueues periodic analyzer jobs (global tenant id). +func EnqueuePostgresAnalyzerJobs(reg *Registry, tenantID string) { + if reg == nil || tenantID == "" { + return + } + kinds := []string{ + KindPostgresMetricsRefresh, + KindPostgresSlowQueryAgg, + KindPostgresTableBloat, + KindPostgresIndexUsage, + KindPostgresAutovacuumLag, + } + for _, k := range kinds { + key := fmt.Sprintf("pgmon-%s-%s", k, tenantID) + idem := key + _, _, _ = reg.Enqueue(tenantID, k, &idem, nil, map[string]any{"trigger": "scheduler"}) + } +} diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index e274445..fcf0790 100644 --- a/internal/jobs/worker.go +++ b/internal/jobs/worker.go @@ -18,6 +18,8 @@ import ( "evobgp/internal/observability" "evobgp/internal/pipeline" "evobgp/internal/store" + + "github.com/jackc/pgx/v5/pgxpool" ) // mergeBirdPostApplyMeta attaches a birdc snapshot after deploy/reload (best-effort). @@ -43,17 +45,28 @@ func mergeBirdPostApplyMeta(j *Job) { } const ( - KindModuleRefresh = "module_refresh" - KindTenantRefresh = "tenant_refresh" - KindPeerReconcile = "peer_reconcile" - KindDeployApply = "deploy_apply" - KindRevisionRollback = "revision_rollback" - KindBirdReload = "bird_reload" + KindModuleRefresh = "module_refresh" + KindTenantRefresh = "tenant_refresh" + KindPeerReconcile = "peer_reconcile" + KindDeployApply = "deploy_apply" + KindRevisionRollback = "revision_rollback" + KindBirdReload = "bird_reload" + KindPostgresMetricsRefresh = "postgres_metrics_refresh" + KindPostgresSlowQueryAgg = "postgres_slow_query_aggregate" + KindPostgresTableBloat = "postgres_table_bloat_estimate" + KindPostgresIndexUsage = "postgres_index_usage_analyze" + KindPostgresAutovacuumLag = "postgres_autovacuum_lag_detect" + KindPostgresVacuum = "postgres_vacuum" + KindPostgresVacuumAnalyze = "postgres_vacuum_analyze" + KindPostgresAnalyze = "postgres_analyze" + KindPostgresReindex = "postgres_reindex" + KindPostgresCleanup = "postgres_cleanup" ) // Worker executes queued jobs against store.Backend (memory or SQL). type Worker struct { Store store.Backend + PgPool *pgxpool.Pool HTTPClient *http.Client // optional; CDN refresh uses this (default 45s timeout). // Registry is set after BootstrapWorkers creates the job queue; used to chain deploy_apply after refresh/rollback. Registry *Registry @@ -155,6 +168,26 @@ func (w *Worker) Process(j *Job) { } mergeBirdPostApplyMeta(j) j.Succeed() + case KindPostgresMetricsRefresh: + w.runPostgresMetricsRefresh(j) + case KindPostgresSlowQueryAgg: + w.runPostgresSlowQueryAgg(j) + case KindPostgresTableBloat: + w.runPostgresTableBloat(j) + case KindPostgresIndexUsage: + w.runPostgresIndexUsage(j) + case KindPostgresAutovacuumLag: + w.runPostgresAutovacuumLag(j) + case KindPostgresVacuum: + w.runPostgresMaint(j, "vacuum") + case KindPostgresVacuumAnalyze: + w.runPostgresMaint(j, "vacuum_analyze") + case KindPostgresAnalyze: + w.runPostgresMaint(j, "analyze") + case KindPostgresReindex: + w.runPostgresMaint(j, "reindex") + case KindPostgresCleanup: + w.runPostgresCleanup(j) default: j.Fail("unknown job kind") } diff --git a/internal/pgmonitor/cache.go b/internal/pgmonitor/cache.go new file mode 100644 index 0000000..05ebd38 --- /dev/null +++ b/internal/pgmonitor/cache.go @@ -0,0 +1,37 @@ +package pgmonitor + +import ( + "sync" + "time" +) + +type cacheEntry struct { + at time.Time + data any +} + +type ttlCache struct { + mu sync.RWMutex + ttl time.Duration + items map[string]cacheEntry +} + +func newTTLCache(ttl time.Duration) *ttlCache { + return &ttlCache{ttl: ttl, items: make(map[string]cacheEntry)} +} + +func (c *ttlCache) get(key string) (any, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + e, ok := c.items[key] + if !ok || time.Since(e.at) > c.ttl { + return nil, false + } + return e.data, true +} + +func (c *ttlCache) set(key string, data any) { + c.mu.Lock() + defer c.mu.Unlock() + c.items[key] = cacheEntry{at: time.Now().UTC(), data: data} +} diff --git a/internal/pgmonitor/correlation.go b/internal/pgmonitor/correlation.go new file mode 100644 index 0000000..5bfc2ee --- /dev/null +++ b/internal/pgmonitor/correlation.go @@ -0,0 +1,83 @@ +package pgmonitor + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Correlation builds aligned timeline points from job_audit and overview cache. +func (s *Service) Correlation(ctx context.Context, windowMinutes int) (CorrelationResponse, error) { + if s == nil || s.pool == nil { + return CorrelationResponse{}, fmt.Errorf("pgmonitor: postgres not configured") + } + if windowMinutes <= 0 { + windowMinutes = 60 + } + if windowMinutes > 1440 { + windowMinutes = 1440 + } + since := time.Now().UTC().Add(-time.Duration(windowMinutes) * time.Minute) + + rows, err := s.pool.Query(ctx, ` + SELECT date_trunc('minute', finished_at) AS bucket, + percentile_cont(0.99) WITHIN GROUP (ORDER BY + EXTRACT(EPOCH FROM (finished_at - started_at)) * 1000) + FROM job_audit + WHERE finished_at >= $1 AND kind IN ('module_refresh', 'tenant_refresh') + AND status = 'succeeded' AND started_at IS NOT NULL + GROUP BY 1 + ORDER BY 1`, since) + if err != nil { + return CorrelationResponse{}, fmt.Errorf("pgmonitor: correlation jobs: %w", err) + } + defer rows.Close() + + points := make(map[time.Time]*CorrelationPoint) + for rows.Next() { + var bucket time.Time + var p99 *float64 + if err := rows.Scan(&bucket, &p99); err != nil { + return CorrelationResponse{}, err + } + bucket = bucket.UTC() + pt := points[bucket] + if pt == nil { + pt = &CorrelationPoint{Timestamp: bucket} + points[bucket] = pt + } + if p99 != nil { + pt.PipelineRefreshP99Ms = *p99 + } + } + + ov, err := s.Overview(ctx) + if err == nil && ov.Database.CacheHitPct > 0 { + now := time.Now().UTC().Truncate(time.Minute) + pt := points[now] + if pt == nil { + pt = &CorrelationPoint{Timestamp: now} + points[now] = pt + } + pt.CacheHitPct = ov.Database.CacheHitPct + } + + out := make([]CorrelationPoint, 0, len(points)) + for _, p := range points { + out = append(out, *p) + } + // simple sort by time + for i := 0; i < len(out); i++ { + for j := i + 1; j < len(out); j++ { + if out[j].Timestamp.Before(out[i].Timestamp) { + out[i], out[j] = out[j], out[i] + } + } + } + return CorrelationResponse{WindowMinutes: windowMinutes, Points: out}, nil +} + +// RecordCorrelationSnapshot is a hook for future Prometheus samples (no-op placeholder). +func RecordCorrelationSnapshot(_ *pgxpool.Pool) {} diff --git a/internal/pgmonitor/maintenance_audit.go b/internal/pgmonitor/maintenance_audit.go new file mode 100644 index 0000000..252c3c6 --- /dev/null +++ b/internal/pgmonitor/maintenance_audit.go @@ -0,0 +1,154 @@ +package pgmonitor + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +// CleanupPolicy names safe retention policies. +type CleanupPolicy string + +const ( + PolicyJobAuditRetention CleanupPolicy = "job_audit_retention" + PolicyASNCacheRetention CleanupPolicy = "asn_cache_retention" +) + +// CleanupRequest for POST /postgres/cleanup. +type CleanupRequest struct { + Policy string `json:"policy"` + DryRun bool `json:"dry_run"` + Limit int `json:"limit"` +} + +// RunCleanup executes a named retention policy. +func RunCleanup(ctx context.Context, pool *pgxpool.Pool, policy string, dryRun bool, limit int) (map[string]any, error) { + if pool == nil { + return nil, fmt.Errorf("pgmonitor: postgres not configured") + } + if limit <= 0 { + limit = 10000 + } + if limit > 100000 { + limit = 100000 + } + detail := map[string]any{"policy": policy, "dry_run": dryRun, "limit": limit} + switch CleanupPolicy(policy) { + case PolicyJobAuditRetention: + cutoff := time.Now().UTC().Add(-90 * 24 * time.Hour) + if dryRun { + var n int64 + err := pool.QueryRow(ctx, ` + SELECT count(*) FROM job_audit + WHERE created_at < $1 AND status IN ('succeeded', 'failed', 'cancelled')`, cutoff).Scan(&n) + detail["would_delete"] = n + return detail, err + } + tag, err := pool.Exec(ctx, ` + DELETE FROM job_audit + WHERE id IN ( + SELECT id FROM job_audit + WHERE created_at < $1 AND status IN ('succeeded', 'failed', 'cancelled') + LIMIT $2 + )`, cutoff, limit) + if err != nil { + return detail, err + } + detail["deleted"] = tag.RowsAffected() + return detail, nil + case PolicyASNCacheRetention: + cutoff := time.Now().UTC().Add(-7 * 24 * time.Hour) + if dryRun { + var n int64 + err := pool.QueryRow(ctx, `SELECT count(*) FROM asn_prefix_cache WHERE fetched_at < $1`, cutoff).Scan(&n) + detail["would_delete"] = n + return detail, err + } + tag, err := pool.Exec(ctx, ` + DELETE FROM asn_prefix_cache WHERE fetched_at < $1`, cutoff) + if err != nil { + return detail, err + } + detail["deleted"] = tag.RowsAffected() + return detail, nil + default: + return nil, fmt.Errorf("pgmonitor: unknown cleanup policy %q", policy) + } +} + +// InsertMaintenanceAudit records an audit row at job start. +func InsertMaintenanceAudit(ctx context.Context, pool *pgxpool.Pool, tenantID, actorPrefix, kind, table string, dryRun bool) (string, error) { + id := uuid.New().String() + _, err := pool.Exec(ctx, ` + INSERT INTO postgres_maintenance_audit + (id, tenant_id, actor_prefix, kind, target_table, dry_run, status, created_at) + VALUES ($1, NULLIF($2,''), NULLIF($3,''), $4, NULLIF($5,''), $6, 'running', now())`, + id, tenantID, actorPrefix, kind, table, dryRun) + return id, err +} + +// FinishMaintenanceAudit updates terminal state. +func FinishMaintenanceAudit(ctx context.Context, pool *pgxpool.Pool, id, status string, detail map[string]any, errMsg *string) error { + var detailJSON []byte + if detail != nil { + detailJSON, _ = json.Marshal(detail) + } + _, err := pool.Exec(ctx, ` + UPDATE postgres_maintenance_audit + SET status = $2, detail_json = $3::jsonb, error_message = $4, + finished_at = now(), started_at = COALESCE(started_at, now()) + WHERE id = $1`, + id, status, string(detailJSON), errMsg) + return err +} + +// ListMaintenanceLogs returns paginated audit rows. +func ListMaintenanceLogs(ctx context.Context, pool *pgxpool.Pool, cursor string, limit int) ([]MaintenanceLogRow, string, bool, error) { + limit = clampLimit(limit, 20, 100) + args := []any{limit + 1} + q := ` + SELECT id, COALESCE(tenant_id,''), COALESCE(actor_prefix,''), kind, + COALESCE(target_table,''), dry_run, status, + detail_json, COALESCE(error_message,''), created_at, started_at, finished_at + FROM postgres_maintenance_audit` + if cursor != "" { + q += ` WHERE created_at < (SELECT created_at FROM postgres_maintenance_audit WHERE id = $2)` + args = append(args, cursor) + } + q += ` ORDER BY created_at DESC LIMIT $1` + + rows, err := pool.Query(ctx, q, args...) + if err != nil { + return nil, "", false, err + } + defer rows.Close() + var out []MaintenanceLogRow + for rows.Next() { + var r MaintenanceLogRow + var detailRaw []byte + var started, finished *time.Time + if err := rows.Scan(&r.ID, &r.TenantID, &r.ActorPrefix, &r.Kind, &r.TargetTable, + &r.DryRun, &r.Status, &detailRaw, &r.Error, &r.CreatedAt, &started, &finished); err != nil { + return nil, "", false, err + } + r.StartedAt = started + r.FinishedAt = finished + if len(detailRaw) > 0 { + _ = json.Unmarshal(detailRaw, &r.Detail) + } + out = append(out, r) + } + hasMore := len(out) > limit + if hasMore { + out = out[:limit] + } + next := "" + if hasMore && len(out) > 0 { + next = out[len(out)-1].ID + } + return out, next, hasMore, rows.Err() +} diff --git a/internal/pgmonitor/queries.go b/internal/pgmonitor/queries.go new file mode 100644 index 0000000..660590b --- /dev/null +++ b/internal/pgmonitor/queries.go @@ -0,0 +1,294 @@ +package pgmonitor + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +func clampLimit(limit, def, max int) int { + if limit <= 0 { + return def + } + if limit > max { + return max + } + return limit +} + +func (s *Service) fetchOverview(ctx context.Context) (Overview, error) { + now := time.Now().UTC() + out := Overview{CollectedAt: now} + + var active, idle, total, maxConn int + err := s.pool.QueryRow(ctx, ` + SELECT + count(*) FILTER (WHERE state = 'active'), + count(*) FILTER (WHERE state = 'idle'), + count(*), + (SELECT setting::int FROM pg_settings WHERE name = 'max_connections') + FROM pg_stat_activity + WHERE datname = current_database()`).Scan(&active, &idle, &total, &maxConn) + if err != nil { + return out, fmt.Errorf("pgmonitor: connections: %w", err) + } + out.Connections = Connections{Active: active, Idle: idle, Total: total, MaxConnections: maxConn} + + var cachePct *float64 + err = s.pool.QueryRow(ctx, ` + SELECT numbackends, xact_commit, xact_rollback, deadlocks, blks_hit, blks_read, + CASE WHEN blks_hit + blks_read > 0 + THEN round(100.0 * blks_hit::numeric / (blks_hit + blks_read), 2) END + FROM pg_stat_database WHERE datname = current_database()`).Scan( + &out.Database.Backends, + &out.Database.XactCommit, + &out.Database.XactRollback, + &out.Database.Deadlocks, + &out.Database.BlksHit, + &out.Database.BlksRead, + &cachePct, + ) + if err != nil { + return out, fmt.Errorf("pgmonitor: database stats: %w", err) + } + if cachePct != nil { + out.Database.CacheHitPct = *cachePct + } + + _ = s.pool.QueryRow(ctx, ` + SELECT checkpoints_timed, checkpoints_req, buffers_checkpoint, buffers_clean, + maxwritten_clean, buffers_backend, buffers_alloc + FROM pg_stat_bgwriter`).Scan( + &out.Bgwriter.CheckpointsTimed, + &out.Bgwriter.CheckpointsReq, + &out.Bgwriter.BuffersCheckpoint, + &out.Bgwriter.BuffersClean, + &out.Bgwriter.MaxWrittenClean, + &out.Bgwriter.BuffersBackend, + &out.Bgwriter.BuffersAlloc, + ) + + _ = s.pool.QueryRow(ctx, `SELECT pg_database_size(current_database())`).Scan(&out.SizeBytes) + + _ = s.pool.QueryRow(ctx, ` + SELECT + (SELECT setting FROM pg_settings WHERE name = 'shared_buffers'), + (SELECT setting FROM pg_settings WHERE name = 'work_mem'), + (SELECT setting FROM pg_settings WHERE name = 'effective_cache_size')`).Scan( + &out.MemorySettings.SharedBuffers, + &out.MemorySettings.WorkMem, + &out.MemorySettings.EffectiveCacheSize, + ) + + rows, err := s.pool.Query(ctx, ` + SELECT client_addr::text, state, sync_state, + EXTRACT(EPOCH FROM COALESCE(write_lag, flush_lag, replay_lag)) * 1000 + FROM pg_stat_replication`) + if err == nil { + defer rows.Close() + for rows.Next() { + var peer ReplicationPeer + var lagMs *float64 + if err := rows.Scan(&peer.ClientAddr, &peer.State, &peer.SyncState, &lagMs); err != nil { + continue + } + if lagMs != nil { + v := int64(*lagMs) + peer.LagMs = &v + } + out.Replication = append(out.Replication, peer) + } + } + + var ext bool + _ = s.pool.QueryRow(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM pg_extension WHERE extname = 'pg_stat_statements')`).Scan(&ext) + out.StatementsEnabled = ext + + return out, nil +} + +func queryLocks(ctx context.Context, pool *pgxpool.Pool) ([]LockRow, error) { + rows, err := pool.Query(ctx, ` + SELECT l.locktype, l.mode, l.granted, a.pid, COALESCE(a.usename, ''), + COALESCE(a.state, ''), COALESCE(left(a.query, 300), ''), + NOT l.granted AS blocked + FROM pg_locks l + JOIN pg_stat_activity a ON a.pid = l.pid + WHERE a.datname = current_database() + AND (NOT l.granted OR l.mode LIKE '%Exclusive%') + ORDER BY l.granted ASC, a.query_start NULLS LAST + LIMIT 200`) + if err != nil { + return nil, fmt.Errorf("pgmonitor: locks: %w", err) + } + defer rows.Close() + var out []LockRow + for rows.Next() { + var r LockRow + if err := rows.Scan(&r.Locktype, &r.Mode, &r.Granted, &r.PID, &r.User, &r.State, &r.Query, &r.Blocked); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +func queryTables(ctx context.Context, pool *pgxpool.Pool, limit int) ([]TableStat, error) { + limit = clampLimit(limit, 20, 100) + rows, err := pool.Query(ctx, ` + SELECT t.relname, + pg_total_relation_size(t.relid), + s.heap_blks_read, s.heap_blks_hit, + t.idx_scan, t.seq_scan, t.n_dead_tup, t.last_autovacuum, + CASE WHEN t.n_live_tup + t.n_dead_tup > 0 + THEN round(t.n_dead_tup::numeric / (t.n_live_tup + t.n_dead_tup), 4) + ELSE 0 END + FROM pg_statio_user_tables s + JOIN pg_stat_user_tables t ON t.relid = s.relid + WHERE t.schemaname = 'public' + ORDER BY pg_total_relation_size(t.relid) DESC + LIMIT $1`, limit) + if err != nil { + return nil, fmt.Errorf("pgmonitor: tables: %w", err) + } + defer rows.Close() + var out []TableStat + for rows.Next() { + var r TableStat + var last *time.Time + if err := rows.Scan(&r.Relname, &r.TotalBytes, &r.HeapBlksRead, &r.HeapBlksHit, + &r.IdxScan, &r.SeqScan, &r.DeadTuples, &last, &r.BloatRatio); err != nil { + return nil, err + } + r.LastAutovacuum = last + out = append(out, r) + } + return out, rows.Err() +} + +// TopQueries loads from pg_stat_statements when available. +func (s *Service) TopQueries(ctx context.Context, limit int) (QueriesResponse, error) { + if s == nil || s.pool == nil { + return QueriesResponse{}, errors.New("pgmonitor: postgres not configured") + } + limit = clampLimit(limit, 20, 100) + now := time.Now().UTC() + + if snap, ok, err := s.loadSnapshot(ctx, "slow_queries", 15*time.Minute); err == nil && ok { + var items []QueryStat + if err := decodePayload(snap.Payload, &items); err == nil { + return QueriesResponse{CollectedAt: snap.CollectedAt, Source: "snapshot", Items: items}, nil + } + } + + if !s.statementsEnabled(ctx) { + return QueriesResponse{CollectedAt: now, Source: "live", Items: nil}, nil + } + items, err := queryTopStatements(ctx, s.pool, limit) + if err != nil { + return QueriesResponse{}, err + } + return QueriesResponse{CollectedAt: now, Source: "live", Items: items}, nil +} + +func (s *Service) statementsEnabled(ctx context.Context) bool { + var ok bool + _ = s.pool.QueryRow(ctx, ` + SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pg_stat_statements')`).Scan(&ok) + return ok +} + +func queryTopStatements(ctx context.Context, pool *pgxpool.Pool, limit int) ([]QueryStat, error) { + rows, err := pool.Query(ctx, ` + SELECT queryid, left(query, 500), calls, total_exec_time, mean_exec_time, rows + FROM pg_stat_statements + WHERE dbid = (SELECT oid FROM pg_database WHERE datname = current_database()) + ORDER BY mean_exec_time DESC + LIMIT $1`, limit) + if err != nil { + if isUndefinedTable(err) { + return nil, nil + } + return nil, fmt.Errorf("pgmonitor: pg_stat_statements: %w", err) + } + defer rows.Close() + var out []QueryStat + for rows.Next() { + var r QueryStat + if err := rows.Scan(&r.QueryID, &r.Query, &r.Calls, &r.TotalExecMs, &r.MeanExecMs, &r.Rows); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +func isUndefinedTable(err error) bool { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + return pgErr.Code == "42P01" || pgErr.Code == "42704" + } + return strings.Contains(err.Error(), "pg_stat_statements") +} + +func isSafeIdent(name string) bool { + if name == "" { + return true + } + for _, r := range name { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' { + continue + } + return false + } + return true +} + +// ExecMaintenance runs VACUUM/ANALYZE/REINDEX with optional dry-run (returns SQL executed or planned). +func ExecMaintenance(ctx context.Context, pool *pgxpool.Pool, kind, table string, dryRun bool) (detail map[string]any, err error) { + if pool == nil { + return nil, errors.New("pgmonitor: postgres not configured") + } + table = strings.TrimSpace(table) + if table != "" && !isSafeIdent(table) { + return nil, errors.New("pgmonitor: invalid table name") + } + qual := "" + if table != "" { + qual = " " + pgx.Identifier{table}.Sanitize() + } + var sql string + switch kind { + case "vacuum": + sql = "VACUUM" + qual + case "vacuum_analyze": + sql = "VACUUM ANALYZE" + qual + case "analyze": + sql = "ANALYZE" + qual + case "reindex": + if table == "" { + return nil, errors.New("pgmonitor: reindex requires table") + } + sql = "REINDEX TABLE" + qual + default: + return nil, fmt.Errorf("pgmonitor: unknown maintenance kind %q", kind) + } + detail = map[string]any{"sql": sql, "dry_run": dryRun} + if dryRun { + return detail, nil + } + _, err = pool.Exec(ctx, sql) + if err != nil { + return detail, fmt.Errorf("pgmonitor: %s: %w", kind, err) + } + detail["executed"] = true + return detail, nil +} diff --git a/internal/pgmonitor/recommendations.go b/internal/pgmonitor/recommendations.go new file mode 100644 index 0000000..e63d861 --- /dev/null +++ b/internal/pgmonitor/recommendations.go @@ -0,0 +1,104 @@ +package pgmonitor + +import ( + "context" + "time" +) + +// Recommendations builds heuristic items from live stats and snapshots. +func (s *Service) Recommendations(ctx context.Context) (RecommendationsResponse, error) { + now := time.Now().UTC() + var items []RecommendationItem + + ov, err := s.Overview(ctx) + if err == nil { + if ov.Database.CacheHitPct > 0 && ov.Database.CacheHitPct < 90 { + items = append(items, RecommendationItem{ + Severity: "warn", + Code: "low_cache_hit", + Title: "Низкий cache hit ratio", + Detail: "Buffer cache hit ниже 90%; проверьте shared_buffers и горячие seq scan.", + }) + } + if ov.Database.Deadlocks > 0 { + items = append(items, RecommendationItem{ + Severity: "warn", + Code: "deadlocks", + Title: "Зафиксированы deadlocks", + Detail: "Проверьте конкурирующие транзакции и порядок блокировок.", + }) + } + if ov.Connections.MaxConnections > 0 && + float64(ov.Connections.Total)/float64(ov.Connections.MaxConnections) > 0.8 { + items = append(items, RecommendationItem{ + Severity: "critical", + Code: "connections_high", + Title: "Много подключений к PostgreSQL", + Detail: "Использование max_connections выше 80%; увеличьте pool tuning или лимит.", + }) + } + } + + tables, err := s.Tables(ctx, 30) + if err == nil { + for _, t := range tables { + if t.SeqScan > 1000 && t.IdxScan < t.SeqScan/10 { + items = append(items, RecommendationItem{ + Severity: "warn", + Code: "missing_index", + Title: "Высокий seq_scan", + Detail: "Таблица часто сканируется последовательно; рассмотрите индекс.", + Refs: []string{t.Relname}, + }) + } + if t.BloatRatio > 0.2 && t.DeadTuples > 5000 { + items = append(items, RecommendationItem{ + Severity: "info", + Code: "autovacuum_lag", + Title: "Возможный bloat / мёртвые строки", + Detail: "Высокая доля n_dead_tup; запланируйте VACUUM.", + Refs: []string{t.Relname}, + }) + } + } + } + + if snap, ok, _ := s.loadSnapshot(ctx, "unused_indexes", 30*time.Minute); ok { + type unused struct { + Index string `json:"index"` + SizeBytes int64 `json:"size_bytes"` + } + var list []unused + if decodePayload(snap.Payload, &list) == nil { + for _, u := range list { + if u.SizeBytes < 1024*1024 { + continue + } + items = append(items, RecommendationItem{ + Severity: "info", + Code: "unused_index", + Title: "Неиспользуемый индекс", + Detail: "idx_scan=0; проверьте перед удалением.", + Refs: []string{u.Index}, + }) + } + } + } + + q, err := s.TopQueries(ctx, 5) + if err == nil { + for _, qs := range q.Items { + if qs.MeanExecMs > 500 { + items = append(items, RecommendationItem{ + Severity: "warn", + Code: "slow_query", + Title: "Медленный запрос", + Detail: "Среднее время выполнения выше 500ms.", + Refs: []string{qs.Query}, + }) + } + } + } + + return RecommendationsResponse{CollectedAt: now, Items: items}, nil +} diff --git a/internal/pgmonitor/scheduler.go b/internal/pgmonitor/scheduler.go new file mode 100644 index 0000000..688ed78 --- /dev/null +++ b/internal/pgmonitor/scheduler.go @@ -0,0 +1,59 @@ +package pgmonitor + +import ( + "context" + "log" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// StartScheduler runs periodic PostgreSQL analyzer snapshots until ctx is cancelled. +func StartScheduler(ctx context.Context, pool *pgxpool.Pool) { + if pool == nil { + return + } + go func() { + t5 := time.NewTicker(5 * time.Minute) + t15 := time.NewTicker(15 * time.Minute) + defer t5.Stop() + defer t15.Stop() + s := NewService(pool) + runLight := func() { + c, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + if err := s.RefreshMetricsSnapshot(c); err != nil { + log.Printf("pgmonitor: metrics refresh: %v", err) + } + if err := s.DetectAutovacuumLag(c); err != nil { + log.Printf("pgmonitor: autovacuum lag: %v", err) + } + } + runHeavy := func() { + c, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + if err := s.AggregateSlowQueries(c, 30); err != nil { + log.Printf("pgmonitor: slow queries: %v", err) + } + if err := s.EstimateTableBloat(c); err != nil { + log.Printf("pgmonitor: bloat: %v", err) + } + if err := s.AnalyzeIndexUsage(c); err != nil { + log.Printf("pgmonitor: index usage: %v", err) + } + } + runLight() + runHeavy() + for { + select { + case <-ctx.Done(): + return + case <-t5.C: + runLight() + case <-t15.C: + runHeavy() + } + } + }() + log.Printf("pgmonitor: scheduler started (5m light / 15m heavy)") +} diff --git a/internal/pgmonitor/service.go b/internal/pgmonitor/service.go new file mode 100644 index 0000000..b5f4fcb --- /dev/null +++ b/internal/pgmonitor/service.go @@ -0,0 +1,90 @@ +package pgmonitor + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Service provides PostgreSQL observability and maintenance helpers (control plane instance scope). +type Service struct { + pool *pgxpool.Pool + cache *ttlCache +} + +// NewService constructs a metrics service for the API PostgreSQL pool. +func NewService(pool *pgxpool.Pool) *Service { + if pool == nil { + return nil + } + return &Service{ + pool: pool, + cache: newTTLCache(10 * time.Second), + } +} + +// Pool exposes the underlying pool for job workers. +func (s *Service) Pool() *pgxpool.Pool { + if s == nil { + return nil + } + return s.pool +} + +// Overview returns cached instance-level stats. +func (s *Service) Overview(ctx context.Context) (Overview, error) { + if s == nil || s.pool == nil { + return Overview{}, errors.New("pgmonitor: postgres not configured") + } + if v, ok := s.cache.get("overview"); ok { + if o, ok := v.(Overview); ok { + return o, nil + } + } + o, err := s.fetchOverview(ctx) + if err != nil { + return Overview{}, err + } + s.cache.set("overview", o) + return o, nil +} + +// Locks returns active / blocking locks. +func (s *Service) Locks(ctx context.Context) ([]LockRow, error) { + if s == nil || s.pool == nil { + return nil, errors.New("pgmonitor: postgres not configured") + } + if v, ok := s.cache.get("locks"); ok { + if rows, ok := v.([]LockRow); ok { + return rows, nil + } + } + rows, err := queryLocks(ctx, s.pool) + if err != nil { + return nil, err + } + s.cache.set("locks", rows) + return rows, nil +} + +// Tables returns top tables by size with I/O stats. +func (s *Service) Tables(ctx context.Context, limit int) ([]TableStat, error) { + if s == nil || s.pool == nil { + return nil, errors.New("pgmonitor: postgres not configured") + } + key := fmt.Sprintf("tables:%d", limit) + if v, ok := s.cache.get(key); ok { + if rows, ok := v.([]TableStat); ok { + return rows, nil + } + } + rows, err := queryTables(ctx, s.pool, limit) + if err != nil { + return nil, err + } + s.cache.set(key, rows) + return rows, nil +} diff --git a/internal/pgmonitor/service_test.go b/internal/pgmonitor/service_test.go new file mode 100644 index 0000000..eeb0405 --- /dev/null +++ b/internal/pgmonitor/service_test.go @@ -0,0 +1,33 @@ +package pgmonitor + +import "testing" + +func TestClampLimit(t *testing.T) { + if clampLimit(0, 20, 100) != 20 { + t.Fatal("default") + } + if clampLimit(200, 20, 100) != 100 { + t.Fatal("max") + } + if clampLimit(5, 20, 100) != 5 { + t.Fatal("value") + } +} + +func TestIsSafeIdent(t *testing.T) { + if !isSafeIdent("revision_materialized_prefix") { + t.Fatal("valid") + } + if isSafeIdent("bad-name") { + t.Fatal("invalid") + } + if !isSafeIdent("") { + t.Fatal("empty ok") + } +} + +func TestNewServiceNilPool(t *testing.T) { + if NewService(nil) != nil { + t.Fatal("expected nil service") + } +} diff --git a/internal/pgmonitor/snapshot.go b/internal/pgmonitor/snapshot.go new file mode 100644 index 0000000..9c78bd1 --- /dev/null +++ b/internal/pgmonitor/snapshot.go @@ -0,0 +1,158 @@ +package pgmonitor + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +type snapshotRow struct { + ID string + CollectedAt time.Time + Payload json.RawMessage +} + +func (s *Service) loadSnapshot(ctx context.Context, id string, maxAge time.Duration) (snapshotRow, bool, error) { + var row snapshotRow + err := s.pool.QueryRow(ctx, ` + SELECT id, collected_at, payload_json + FROM postgres_monitor_snapshot + WHERE id = $1 AND collected_at >= $2`, + id, time.Now().UTC().Add(-maxAge)).Scan(&row.ID, &row.CollectedAt, &row.Payload) + if err != nil { + return snapshotRow{}, false, nil + } + return row, true, nil +} + +func (s *Service) UpsertSnapshot(ctx context.Context, id string, payload any) error { + if s == nil || s.pool == nil { + return fmt.Errorf("pgmonitor: postgres not configured") + } + b, err := json.Marshal(payload) + if err != nil { + return err + } + _, err = s.pool.Exec(ctx, ` + INSERT INTO postgres_monitor_snapshot (id, collected_at, payload_json) + VALUES ($1, now(), $2::jsonb) + ON CONFLICT (id) DO UPDATE SET collected_at = EXCLUDED.collected_at, payload_json = EXCLUDED.payload_json`, + id, string(b)) + return err +} + +func decodePayload(raw json.RawMessage, dest any) error { + return json.Unmarshal(raw, dest) +} + +// RefreshMetricsSnapshot stores overview and tables for heavy reads. +func (s *Service) RefreshMetricsSnapshot(ctx context.Context) error { + ov, err := s.fetchOverview(ctx) + if err != nil { + return err + } + if err := s.UpsertSnapshot(ctx, "overview", ov); err != nil { + return err + } + tables, err := queryTables(ctx, s.pool, 50) + if err != nil { + return err + } + return s.UpsertSnapshot(ctx, "tables", tables) +} + +// AggregateSlowQueries stores top statements snapshot. +func (s *Service) AggregateSlowQueries(ctx context.Context, limit int) error { + if !s.statementsEnabled(ctx) { + return s.UpsertSnapshot(ctx, "slow_queries", []QueryStat{}) + } + items, err := queryTopStatements(ctx, s.pool, clampLimit(limit, 20, 100)) + if err != nil { + return err + } + return s.UpsertSnapshot(ctx, "slow_queries", items) +} + +// EstimateTableBloat refreshes bloat heuristics on tables snapshot. +func (s *Service) EstimateTableBloat(ctx context.Context) error { + tables, err := queryTables(ctx, s.pool, 100) + if err != nil { + return err + } + return s.UpsertSnapshot(ctx, "table_bloat", tables) +} + +// AnalyzeIndexUsage stores unused indexes. +func (s *Service) AnalyzeIndexUsage(ctx context.Context) error { + rows, err := s.pool.Query(ctx, ` + SELECT indexrelname, idx_scan, pg_relation_size(indexrelid) + FROM pg_stat_user_indexes + WHERE schemaname = 'public' AND idx_scan = 0 + ORDER BY pg_relation_size(indexrelid) DESC + LIMIT 50`) + if err != nil { + return fmt.Errorf("pgmonitor: index usage: %w", err) + } + defer rows.Close() + type unused struct { + Index string `json:"index"` + IdxScan int64 `json:"idx_scan"` + SizeBytes int64 `json:"size_bytes"` + } + var items []unused + for rows.Next() { + var u unused + if err := rows.Scan(&u.Index, &u.IdxScan, &u.SizeBytes); err != nil { + return err + } + items = append(items, u) + } + return s.UpsertSnapshot(ctx, "unused_indexes", items) +} + +// DetectAutovacuumLag stores tables with high dead tuple ratio. +func (s *Service) DetectAutovacuumLag(ctx context.Context) error { + rows, err := s.pool.Query(ctx, ` + SELECT relname, n_dead_tup, last_autovacuum, + CASE WHEN n_live_tup + n_dead_tup > 0 + THEN round(n_dead_tup::numeric / (n_live_tup + n_dead_tup), 4) ELSE 0 END + FROM pg_stat_user_tables + WHERE schemaname = 'public' AND n_dead_tup > 1000 + ORDER BY n_dead_tup DESC + LIMIT 30`) + if err != nil { + return fmt.Errorf("pgmonitor: autovacuum lag: %w", err) + } + defer rows.Close() + type lagRow struct { + Relname string `json:"relname"` + DeadTuples int64 `json:"n_dead_tup"` + LastAutovacuum *time.Time `json:"last_autovacuum,omitempty"` + Ratio float64 `json:"ratio"` + } + var items []lagRow + for rows.Next() { + var r lagRow + if err := rows.Scan(&r.Relname, &r.DeadTuples, &r.LastAutovacuum, &r.Ratio); err != nil { + return err + } + items = append(items, r) + } + return s.UpsertSnapshot(ctx, "autovacuum_lag", items) +} + +// RunPeriodicAnalyzerJobs runs all snapshot analyzers (for scheduler). +func RunPeriodicAnalyzerJobs(ctx context.Context, pool *pgxpool.Pool) { + s := NewService(pool) + if s == nil { + return + } + _ = s.RefreshMetricsSnapshot(ctx) + _ = s.AggregateSlowQueries(ctx, 30) + _ = s.EstimateTableBloat(ctx) + _ = s.AnalyzeIndexUsage(ctx) + _ = s.DetectAutovacuumLag(ctx) +} diff --git a/internal/pgmonitor/types.go b/internal/pgmonitor/types.go new file mode 100644 index 0000000..642dbab --- /dev/null +++ b/internal/pgmonitor/types.go @@ -0,0 +1,148 @@ +package pgmonitor + +import "time" + +// Overview is instance-level PostgreSQL health snapshot. +type Overview struct { + CollectedAt time.Time `json:"collected_at"` + Connections Connections `json:"connections"` + Database DatabaseStats `json:"database"` + Bgwriter BgwriterStats `json:"bgwriter"` + SizeBytes int64 `json:"database_size_bytes"` + MemorySettings MemorySettings `json:"memory_settings"` + Replication []ReplicationPeer `json:"replication"` + StatementsEnabled bool `json:"pg_stat_statements_enabled"` +} + +// Connections summarizes pg_stat_activity for current database. +type Connections struct { + Active int `json:"active"` + Idle int `json:"idle"` + Total int `json:"total"` + MaxConnections int `json:"max_connections"` +} + +// DatabaseStats from pg_stat_database. +type DatabaseStats struct { + Backends int `json:"backends"` + XactCommit int64 `json:"xact_commit"` + XactRollback int64 `json:"xact_rollback"` + Deadlocks int64 `json:"deadlocks"` + BlksHit int64 `json:"blks_hit"` + BlksRead int64 `json:"blks_read"` + CacheHitPct float64 `json:"cache_hit_pct"` +} + +// BgwriterStats from pg_stat_bgwriter. +type BgwriterStats struct { + CheckpointsTimed int64 `json:"checkpoints_timed"` + CheckpointsReq int64 `json:"checkpoints_req"` + BuffersCheckpoint int64 `json:"buffers_checkpoint"` + BuffersClean int64 `json:"buffers_clean"` + MaxWrittenClean int64 `json:"maxwritten_clean"` + BuffersBackend int64 `json:"buffers_backend"` + BuffersAlloc int64 `json:"buffers_alloc"` +} + +// MemorySettings is best-effort from pg_settings (not RSS). +type MemorySettings struct { + SharedBuffers string `json:"shared_buffers"` + WorkMem string `json:"work_mem"` + EffectiveCacheSize string `json:"effective_cache_size"` +} + +// ReplicationPeer from pg_stat_replication. +type ReplicationPeer struct { + ClientAddr string `json:"client_addr,omitempty"` + State string `json:"state"` + SyncState string `json:"sync_state,omitempty"` + LagMs *int64 `json:"lag_ms,omitempty"` +} + +// QueryStat is a row from pg_stat_statements or snapshot. +type QueryStat struct { + QueryID int64 `json:"queryid,omitempty"` + Query string `json:"query"` + Calls int64 `json:"calls"` + TotalExecMs float64 `json:"total_exec_ms"` + MeanExecMs float64 `json:"mean_exec_ms"` + Rows int64 `json:"rows"` +} + +// QueriesResponse for GET /monitoring/postgres/queries. +type QueriesResponse struct { + CollectedAt time.Time `json:"collected_at"` + Source string `json:"source"` // live | snapshot + Items []QueryStat `json:"items"` +} + +// LockRow describes a lock / blocking session. +type LockRow struct { + Locktype string `json:"locktype"` + Mode string `json:"mode"` + Granted bool `json:"granted"` + PID int32 `json:"pid"` + User string `json:"usename,omitempty"` + State string `json:"state,omitempty"` + Query string `json:"query,omitempty"` + Blocked bool `json:"blocked"` +} + +// TableStat combines size and scan stats for a user table. +type TableStat struct { + Relname string `json:"relname"` + TotalBytes int64 `json:"total_bytes"` + HeapBlksRead int64 `json:"heap_blks_read"` + HeapBlksHit int64 `json:"heap_blks_hit"` + IdxScan int64 `json:"idx_scan"` + SeqScan int64 `json:"seq_scan"` + DeadTuples int64 `json:"n_dead_tup"` + LastAutovacuum *time.Time `json:"last_autovacuum,omitempty"` + BloatRatio float64 `json:"bloat_ratio,omitempty"` +} + +// RecommendationItem is a heuristic ops hint. +type RecommendationItem struct { + Severity string `json:"severity"` // info | warn | critical + Code string `json:"code"` + Title string `json:"title"` + Detail string `json:"detail"` + Refs []string `json:"refs,omitempty"` +} + +// RecommendationsResponse for GET recommendations. +type RecommendationsResponse struct { + CollectedAt time.Time `json:"collected_at"` + Items []RecommendationItem `json:"items"` +} + +// CorrelationPoint is one aligned sample for overlay charts. +type CorrelationPoint struct { + Timestamp time.Time `json:"timestamp"` + PipelineRefreshP99Ms float64 `json:"pipeline_refresh_p99_ms,omitempty"` + BirdScrapeOK *float64 `json:"bird_scrape_ok,omitempty"` + HTTPRequestRate float64 `json:"http_request_rate,omitempty"` + CacheHitPct float64 `json:"cache_hit_pct,omitempty"` +} + +// CorrelationResponse for GET /monitoring/correlation. +type CorrelationResponse struct { + WindowMinutes int `json:"window_minutes"` + Points []CorrelationPoint `json:"points"` +} + +// MaintenanceLogRow is an audit entry. +type MaintenanceLogRow struct { + ID string `json:"id"` + TenantID string `json:"tenant_id,omitempty"` + ActorPrefix string `json:"actor_prefix,omitempty"` + Kind string `json:"kind"` + TargetTable string `json:"target_table,omitempty"` + DryRun bool `json:"dry_run"` + Status string `json:"status"` + Detail map[string]any `json:"detail,omitempty"` + Error string `json:"error,omitempty"` + CreatedAt time.Time `json:"created_at"` + StartedAt *time.Time `json:"started_at,omitempty"` + FinishedAt *time.Time `json:"finished_at,omitempty"` +} diff --git a/migrations/postgres/000023_pg_stat_statements.down.sql b/migrations/postgres/000023_pg_stat_statements.down.sql new file mode 100644 index 0000000..be80e5c --- /dev/null +++ b/migrations/postgres/000023_pg_stat_statements.down.sql @@ -0,0 +1 @@ +DROP EXTENSION IF EXISTS pg_stat_statements; diff --git a/migrations/postgres/000023_pg_stat_statements.up.sql b/migrations/postgres/000023_pg_stat_statements.up.sql new file mode 100644 index 0000000..f0af5ad --- /dev/null +++ b/migrations/postgres/000023_pg_stat_statements.up.sql @@ -0,0 +1,2 @@ +-- pg_stat_statements requires shared_preload_libraries on the server; extension may fail on dev without restart. +CREATE EXTENSION IF NOT EXISTS pg_stat_statements; diff --git a/migrations/postgres/000024_postgres_monitoring.down.sql b/migrations/postgres/000024_postgres_monitoring.down.sql new file mode 100644 index 0000000..3d62cb1 --- /dev/null +++ b/migrations/postgres/000024_postgres_monitoring.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS postgres_maintenance_audit; +DROP TABLE IF EXISTS postgres_monitor_snapshot; diff --git a/migrations/postgres/000024_postgres_monitoring.up.sql b/migrations/postgres/000024_postgres_monitoring.up.sql new file mode 100644 index 0000000..a473d62 --- /dev/null +++ b/migrations/postgres/000024_postgres_monitoring.up.sql @@ -0,0 +1,29 @@ +CREATE TABLE IF NOT EXISTS postgres_monitor_snapshot ( + id TEXT PRIMARY KEY, + collected_at TIMESTAMPTZ NOT NULL, + payload_json JSONB NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_postgres_monitor_snapshot_collected + ON postgres_monitor_snapshot (collected_at DESC); + +CREATE TABLE IF NOT EXISTS postgres_maintenance_audit ( + id TEXT PRIMARY KEY, + tenant_id TEXT, + actor_prefix TEXT, + kind TEXT NOT NULL, + target_table TEXT, + dry_run BOOLEAN NOT NULL DEFAULT false, + status TEXT NOT NULL, + detail_json JSONB, + error_message TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ, + CONSTRAINT postgres_maintenance_audit_status_chk CHECK ( + status IN ('queued', 'running', 'succeeded', 'failed', 'cancelled') + ) +); + +CREATE INDEX IF NOT EXISTS idx_postgres_maintenance_audit_created + ON postgres_maintenance_audit (created_at DESC); diff --git a/migrations/sqlite/000023_pg_stat_statements.down.sql b/migrations/sqlite/000023_pg_stat_statements.down.sql new file mode 100644 index 0000000..9f7060a --- /dev/null +++ b/migrations/sqlite/000023_pg_stat_statements.down.sql @@ -0,0 +1 @@ +-- no-op diff --git a/migrations/sqlite/000023_pg_stat_statements.up.sql b/migrations/sqlite/000023_pg_stat_statements.up.sql new file mode 100644 index 0000000..99c6d8e --- /dev/null +++ b/migrations/sqlite/000023_pg_stat_statements.up.sql @@ -0,0 +1 @@ +-- no-op: pg_stat_statements is PostgreSQL-only diff --git a/migrations/sqlite/000024_postgres_monitoring.down.sql b/migrations/sqlite/000024_postgres_monitoring.down.sql new file mode 100644 index 0000000..3d62cb1 --- /dev/null +++ b/migrations/sqlite/000024_postgres_monitoring.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS postgres_maintenance_audit; +DROP TABLE IF EXISTS postgres_monitor_snapshot; diff --git a/migrations/sqlite/000024_postgres_monitoring.up.sql b/migrations/sqlite/000024_postgres_monitoring.up.sql new file mode 100644 index 0000000..e3d7a31 --- /dev/null +++ b/migrations/sqlite/000024_postgres_monitoring.up.sql @@ -0,0 +1,20 @@ +CREATE TABLE IF NOT EXISTS postgres_monitor_snapshot ( + id TEXT PRIMARY KEY, + collected_at TEXT NOT NULL, + payload_json TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS postgres_maintenance_audit ( + id TEXT PRIMARY KEY, + tenant_id TEXT, + actor_prefix TEXT, + kind TEXT NOT NULL, + target_table TEXT, + dry_run INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL, + detail_json TEXT, + error_message TEXT, + created_at TEXT NOT NULL, + started_at TEXT, + finished_at TEXT +); diff --git a/web/src/lib/components/monitoring/MonitoringPostgresTab.svelte b/web/src/lib/components/monitoring/MonitoringPostgresTab.svelte new file mode 100644 index 0000000..a3b12e2 --- /dev/null +++ b/web/src/lib/components/monitoring/MonitoringPostgresTab.svelte @@ -0,0 +1,500 @@ + + +
+
+ + Instance-level PostgreSQL (control plane) +
+
+
+ + +
+ +
+
+ +{#if unavailable} + + PostgreSQL недоступен + + Мониторинг требует EVOBGP_DATABASE_URL (не memory backend). + + +{:else} + + + Обзор + Запросы + Блокировки + Таблицы + Обслуживание + Корреляция + + + + {#if overview} +
+ + + Подключения + + +

+ {overview.connections.active} / {overview.connections.max_connections} +

+
+
+
+

+ idle {overview.connections.idle}, total {overview.connections.total} +

+ + + + + Cache hit + + +

+ {overview.database.cache_hit_pct ?? '—'}% +

+
+
+
+ + + + + TPS (commits) + + +

+ {overview.database.xact_commit.toLocaleString()} +

+

+ rollback {overview.database.xact_rollback.toLocaleString()}, deadlocks {overview + .database.deadlocks} +

+ + + + + Размер БД + + +

{formatBytes(overview.database_size_bytes)}

+

+ shared_buffers {overview.memory_settings.shared_buffers} +

+ + +
+ {#if overview.replication?.length} + + + Репликация + + + + + + Адрес + Состояние + Lag ms + + + + {#each overview.replication as r (r.client_addr ?? r.state)} + + {r.client_addr ?? '—'} + {r.state} + {r.lag_ms ?? '—'} + + {/each} + +
+ + + {/if} + {/if} + + + + + + Медленные запросы + + Источник: {queries?.source ?? '—'} + {#if overview && !overview.pg_stat_statements_enabled} + · pg_stat_statements не включён + {/if} + + + + + + + mean ms + calls + query + + + + {#each queries?.items ?? [] as q (q.queryid ?? q.query)} + + {q.mean_exec_ms.toFixed(1)} + {q.calls} + {q.query} + + {:else} + + Нет данных + + {/each} + +
+ + + + + + + + Блокировки + + + + + + pid + mode + granted + query + + + + {#each locks as l (l.pid)} + + {l.pid} + + {l.mode} + + {l.granted ? 'да' : 'нет'} + {l.query ?? '—'} + + {:else} + + Нет активных блокировок + + {/each} + +
+ + + + + + + + Таблицы и хранилище + + + + + + table + size + seq_scan + idx_scan + bloat + + + + {#each tables as t (t.relname)} + + {t.relname} + {formatBytes(t.total_bytes)} + {t.seq_scan} + {t.idx_scan} + {(t.bloat_ratio ?? 0).toFixed(2)} + + {/each} + +
+ + + {#if recommendations?.items?.length} + + + Рекомендации + + + {#each recommendations.items as item (item.code + item.title)} + + {item.title} + {item.detail} + + {/each} + + + {/if} + + + + {#if !isOperator} + + Только operator + Обслуживание БД доступно с ролью operator. + + {:else} + + + Операции + Все операции — async job (202). По умолчанию dry-run. + + + + + + + + + {/if} + + + Журнал обслуживания + + + + + + время + kind + status + dry_run + + + + {#each maintLogs as log (log.id)} + + {log.created_at} + {log.kind} + {log.status} + {log.dry_run ? 'да' : 'нет'} + + {:else} + + Пусто + + {/each} + +
+ + + + + + + + Корреляция (1ч) + Pipeline refresh p99 vs cache hit по минутам + + + {#each correlation?.points ?? [] as p (p.timestamp)} +
+ {p.timestamp} + p99 refresh: {p.pipeline_refresh_p99_ms?.toFixed(0) ?? '—'} ms + cache hit: {p.cache_hit_pct?.toFixed(1) ?? '—'}% +
+ {:else} +

Нет точек за окно

+ {/each} + + + + +{/if} diff --git a/web/src/lib/monitoring/postgres.ts b/web/src/lib/monitoring/postgres.ts new file mode 100644 index 0000000..82ab7ad --- /dev/null +++ b/web/src/lib/monitoring/postgres.ts @@ -0,0 +1,116 @@ +/** Types and helpers for PostgreSQL monitoring API. */ + +export const POSTGRES_POLL_MS = 20_000; +export const POSTGRES_SLOW_POLL_MS = 60_000; + +export type PostgresOverview = { + collected_at: string; + connections: { + active: number; + idle: number; + total: number; + max_connections: number; + }; + database: { + backends: number; + xact_commit: number; + xact_rollback: number; + deadlocks: number; + blks_hit: number; + blks_read: number; + cache_hit_pct: number; + }; + database_size_bytes: number; + memory_settings: { + shared_buffers: string; + work_mem: string; + effective_cache_size: string; + }; + replication: Array<{ + client_addr?: string; + state: string; + sync_state?: string; + lag_ms?: number; + }>; + pg_stat_statements_enabled: boolean; +}; + +export type PostgresQueryRow = { + queryid?: number; + query: string; + calls: number; + total_exec_ms: number; + mean_exec_ms: number; + rows: number; +}; + +export type PostgresQueriesResponse = { + collected_at: string; + source: string; + items: PostgresQueryRow[]; +}; + +export type PostgresLockRow = { + locktype: string; + mode: string; + granted: boolean; + pid: number; + usename?: string; + state?: string; + query?: string; + blocked: boolean; +}; + +export type PostgresTableRow = { + relname: string; + total_bytes: number; + idx_scan: number; + seq_scan: number; + n_dead_tup: number; + bloat_ratio?: number; + last_autovacuum?: string; +}; + +export type PostgresRecommendation = { + severity: string; + code: string; + title: string; + detail: string; + refs?: string[]; +}; + +export type PostgresRecommendationsResponse = { + collected_at: string; + items: PostgresRecommendation[]; +}; + +export type PostgresMaintLog = { + id: string; + kind: string; + target_table?: string; + dry_run: boolean; + status: string; + error?: string; + created_at: string; +}; + +export type CorrelationResponse = { + window_minutes: number; + points: Array<{ + timestamp: string; + pipeline_refresh_p99_ms?: number; + cache_hit_pct?: number; + }>; +}; + +export function formatBytes(n: number): string { + if (n >= 1 << 30) return `${(n / (1 << 30)).toFixed(1)} GiB`; + if (n >= 1 << 20) return `${(n / (1 << 20)).toFixed(1)} MiB`; + if (n >= 1 << 10) return `${(n / (1 << 10)).toFixed(1)} KiB`; + return `${n} B`; +} + +export function connUsagePct(ov: PostgresOverview | null): number { + if (!ov?.connections.max_connections) return 0; + return Math.min(100, (ov.connections.total / ov.connections.max_connections) * 100); +} diff --git a/web/src/routes/monitoring/+page.svelte b/web/src/routes/monitoring/+page.svelte index cc6846d..a044fcc 100644 --- a/web/src/routes/monitoring/+page.svelte +++ b/web/src/routes/monitoring/+page.svelte @@ -49,6 +49,8 @@ import KpiMetricsGrid from '$lib/ui/patterns/kpi/kpi-metrics-grid.svelte'; import ScrollPreBlock from '$lib/components/app/scroll-pre-block.svelte'; import PageHeader from '$lib/ui/app/page-header/page-header.svelte'; + import MonitoringPostgresTab from '$lib/components/monitoring/MonitoringPostgresTab.svelte'; + import { Tabs, TabsContent, TabsList, TabsTrigger } from '$lib/ui/core/tabs/index.js'; import { cn } from '$lib/utils.js'; import RefreshCw from '@lucide/svelte/icons/refresh-cw'; import Gauge from '@lucide/svelte/icons/gauge'; @@ -80,6 +82,7 @@ let lastUpdated = $state(null); let initialLoading = $state(true); let refreshing = $state(false); + let mainTab = $state('system'); const statAccents = [ { @@ -322,337 +325,362 @@ {/snippet} - {#if !initialLoading} - {#if overallStatus === 'ok'} - - - Система в норме - {overallHint} - - {:else if overallStatus === 'warn'} - - - Требуется внимание - {overallHint} - - {:else if overallStatus === 'error'} - - - Обнаружена проблема - {overallHint} - - {/if} - {/if} + + + Система + PostgreSQL + - + + {#if !initialLoading} + {#if overallStatus === 'ok'} + + + Система в норме + {overallHint} + + {:else if overallStatus === 'warn'} + + + Требуется внимание + {overallHint} + + {:else if overallStatus === 'error'} + + + Обнаружена проблема + {overallHint} + + {/if} + {/if} -
- {#if initialLoading} - - - {:else} - - - Доступность и готовность - GET /v1/health · GET /v1/ready - - - {#if health?.error || readyError} - - - Ошибка проверки - - {#if health?.error}{health.error}{/if} - {#if health?.error && readyError}
{/if} - {#if readyError}{readyError}{/if} - - - {/if} + - - - - Проверка - Статус - - - - {@const liveBadge = livenessBadge(health)} - - -
-
- - - {liveBadge.label} - - - {@const readyBadge = readinessBadge(ready)} - - -
-
- - - {readyBadge.label} - - - {#if ready?.checks && Object.keys(ready.checks).length > 0} - - -

Зависимости

- - - {#each Object.entries(ready.checks) as [key, value] (key)} - {@const badge = checkStatusBadge(value)} - {@const CheckIcon = checkIconByKey[key] ?? ListTodo} +
+ {#if initialLoading} + + + {:else} + + + Доступность и готовность + GET /v1/health · GET /v1/ready + + + {#if health?.error || readyError} + + + Ошибка проверки + + {#if health?.error}{health.error}{/if} + {#if health?.error && readyError}
{/if} + {#if readyError}{readyError}{/if} + + + {/if} + +
+ + + Проверка + Статус + + + + {@const liveBadge = livenessBadge(health)}
-
-
- {badge.label} - {#if badge.hint} -

{badge.hint}

- {/if} -
+ {liveBadge.label} - {/each} - {/if} - -
- -

- HTTP 503 на readiness означает недоступность одной из зависимостей в checks. -

- - - - - -
-
- - - BGP на API-хосте - - GET /v1/bird/status -
- -
- - - {#if birdError} - - - Ошибка birdc - {birdError} - - {:else if bird && !bird.birdc_configured} -

- {bird.message ?? 'birdc не настроен на API-хосте (EVOBGP_BIRDC_SOCKET).'} -

- {:else if bird} -
-
- Established / total - - {bird.bgp_established} / {bird.bgp_sessions_total} - {#if bgpRatio !== null} - ({bgpRatio}%) - {/if} - -
- {#if bgpRatio !== null} -
-
= 100 - ? 'bg-success' - : bgpRatio >= 50 - ? 'bg-warning' - : 'bg-destructive' - )} - style="width: {bgpRatio}%" - >
-
- {/if} - {#if bird.error} -

{bird.error}

- {/if} -
- {#if bird.protocols_excerpt} - -
-

Вывод birdc (protocols)

- -
- {/if} - {/if} - - - {/if} -
- -
- {#if initialLoading} - - - {:else} - - -
-
- - - Задачи - - Последние 100 задач · GET /v1/jobs -
- -
- - - {#if jobsError} - - - Не удалось загрузить задачи - {jobsError} - - {:else if jobs} -
-
-

Активных

-

{jobs.running}

-
-
-

С ошибками

-

0 ? 'text-warning' : 'text-success' - )} - > - {jobs.failed} -

-
-
-

В выборке

-

{jobs.total}

-
-
- - - - {#if failedJobs.length > 0} -
-

Последние ошибки

-
    - {#each failedJobs as job (job.job_id)} -
  • -
    -

    {jobKindTitle(job)}

    - {jobStatusRu(job.status)} + {@const readyBadge = readinessBadge(ready)} + + +
    +
    - {#if job.error} -

    - {truncateError(job.error)} -

    - {/if} -
    • - {/each} -
-
- {:else} -

- Критичных сбоев в последних 100 задачах нет. -

- {/if} - {/if} - - + + + {readyBadge.label} + + + {#if ready?.checks && Object.keys(ready.checks).length > 0} + + +

Зависимости

+ + + {#each Object.entries(ready.checks) as [key, value] (key)} + {@const badge = checkStatusBadge(value)} + {@const CheckIcon = checkIconByKey[key] ?? ListTodo} + + +
+
+ + +
+ {badge.label} + {#if badge.hint} +

{badge.hint}

+ {/if} +
+ + + {/each} + {/if} + + - - - - - Что проверять при деградации - - Короткая шпаргалка для triage - - - - - API недоступен - - Если /v1/health возвращает ошибку — проверьте процесс API и - его логи. - - - - - Readiness не «Готов» - - Сначала postgres, затем - store и jobs в checks. - - - - - Низкий ratio BGP - - Проверьте /v1/bird/status, затем состояние пиров в - . - - - - - Ошибки задач - - Откройте - - и проверьте последние неуспешные jobs. - - - - - {/if} -
+

+ HTTP 503 на readiness означает недоступность одной из зависимостей в checks. +

+ + + + + +
+
+ + + BGP на API-хосте + + GET /v1/bird/status +
+ +
+ + + {#if birdError} + + + Ошибка birdc + {birdError} + + {:else if bird && !bird.birdc_configured} +

+ {bird.message ?? 'birdc не настроен на API-хосте (EVOBGP_BIRDC_SOCKET).'} +

+ {:else if bird} +
+
+ Established / total + + {bird.bgp_established} / {bird.bgp_sessions_total} + {#if bgpRatio !== null} + ({bgpRatio}%) + {/if} + +
+ {#if bgpRatio !== null} +
+
= 100 + ? 'bg-success' + : bgpRatio >= 50 + ? 'bg-warning' + : 'bg-destructive' + )} + style="width: {bgpRatio}%" + >
+
+ {/if} + {#if bird.error} +

{bird.error}

+ {/if} +
+ {#if bird.protocols_excerpt} + +
+

Вывод birdc (protocols)

+ +
+ {/if} + {/if} + + + {/if} + + +
+ {#if initialLoading} + + + {:else} + + +
+
+ + + Задачи + + Последние 100 задач · GET /v1/jobs +
+ +
+ + + {#if jobsError} + + + Не удалось загрузить задачи + {jobsError} + + {:else if jobs} +
+
+

Активных

+

{jobs.running}

+
+
+

С ошибками

+

0 ? 'text-warning' : 'text-success' + )} + > + {jobs.failed} +

+
+
+

В выборке

+

{jobs.total}

+
+
+ + + + {#if failedJobs.length > 0} +
+

Последние ошибки

+
    + {#each failedJobs as job (job.job_id)} +
  • +
    +

    {jobKindTitle(job)}

    + {jobStatusRu(job.status)} +
    + {#if job.error} +

    + {truncateError(job.error)} +

    + {/if} +
    • + {/each} +
+
+ {:else} +

+ Критичных сбоев в последних 100 задачах нет. +

+ {/if} + {/if} + + + + + + + + Что проверять при деградации + + Короткая шпаргалка для triage + + + + + API недоступен + + Если /v1/health возвращает ошибку — проверьте процесс API + и его логи. + + + + + Readiness не «Готов» + + Сначала postgres, затем + store и jobs в checks. + + + + + Низкий ratio BGP + + Проверьте /v1/bird/status, затем состояние пиров в + . + + + + + Ошибки задач + + Откройте + + и проверьте последние неуспешные jobs. + + + + + {/if} +
+ + + + + +