diff --git a/README.md b/README.md index e1d8026..54f4d45 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,8 @@ docker compose logs -f gateway |----------|------------| | **[docs/GATEWAY_RUN.md](docs/GATEWAY_RUN.md)** | Полная инструкция: конфиг, pull/registry, Docker CLI, Compose, CI/CD, неполадки | | **[docs/API.md](docs/API.md)** | Контракт Telemt Control API (`/v1/…`) | -| **[docs/AGGREGATE.md](docs/AGGREGATE.md)** | Агрегирующие эндпоинты шлюза (`/api/agg/…`) | +| **[docs/AGGREGATE.md](docs/AGGREGATE.md)** | Агрегирующие эндпоинты шлюза (`/api/agg/…`), CORS, кэш | +| **[docs/AGGREGATE_OPENAPI.yaml](docs/AGGREGATE_OPENAPI.yaml)** | OpenAPI 3 черновик для `/api/agg/*` (генерация типов для UI) | | **[docs/GEOIP.md](docs/GEOIP.md)** | GeoLite2 City (страна/город) и опционально ASN (номер AS, организация) для IP в `unique-ips` | ## Сборка и тесты без Docker diff --git a/config.example.yaml b/config.example.yaml index ac071a0..4629013 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -23,11 +23,17 @@ whitelist_cidrs: # - "10.0.0.0/8" trusted_proxies: [] -# Опционально: по умолчанию /api/agg/* опрашивает все servers; можно ограничить список: +# SPA на другом origin (preflight OPTIONS + Access-Control-Allow-Origin): +# cors_allowed_origins: +# - "http://localhost:5173" +# # - "*" + +# Опционально: по умолчанию /api/agg/* опрашивает все servers; можно ограничить список и включить кэш ответов: # aggregate: # include_aliases: # - gt1 # - gt2 +# cache_ttl_ms: 2000 # Геолокация IP в /api/agg/unique-ips. См. docs/GEOIP.md (City + опционально ASN; Country-only не нужен) # geoip: diff --git a/docs/AGGREGATE.md b/docs/AGGREGATE.md index 071816d..f715192 100644 --- a/docs/AGGREGATE.md +++ b/docs/AGGREGATE.md @@ -1,11 +1,28 @@ # Агрегирующие эндпоинты шлюза (`/api/agg/`) -Шлюз **telemt-api** опрашивает несколько upstream [Telemt Control API](API.md) (`GET /v1/stats/users` на каждом сервере из конфигурации) и отдаёт сводные JSON-ответы в формате `{"ok": true, "data": ...}`. +Шлюз **telemt-api** опрашивает несколько upstream [Telemt Control API](API.md) и отдаёт сводные JSON-ответы. + +- Большинство маршрутов агрегации используют **`GET /v1/stats/users`** на каждом сервере из конфигурации. +- **`GET /api/agg/fleet-status`** дополнительно вызывает на каждом upstream **`GET /v1/health`** и **`GET /v1/system/info`** (параллельно по серверам). **Единицы трафика в агрегатах:** поля `*_megabytes` — это **двоичные мегабайты (MiB)**, 1 MiB = 1024² октетов (как у Telemt в ответе считаются октеты, шлюз делит на MiB для удобства). Доступ к **одному** инстансу по-прежнему через прокси: `GET /api/{alias}/…` (например `/api/gt1/v1/stats/users`) — там по-прежнему `total_octets` как в [API.md](API.md). +## Успешный ответ (общий контракт) + +```json +{ + "ok": true, + "data": {}, + "generated_at": "2026-03-30T12:00:00.000000000Z", + "partial": true +} +``` + +- **`generated_at`** — UTC, RFC3339Nano; время формирования ответа шлюза. +- **`partial`** — присутствует и равно `true`, если **хотя бы один** upstream в этом запросе завершился с ошибкой (HTTP не 200, сеть, `ok: false` в теле и т.д.), либо для `fleet-status` — если не оба подзапроса (health и system/info) успешны для какой-либо ноды. Если все вызовы успешны, поле **`partial` не включается**. + ## Маршруты | Метод | Путь | Описание | @@ -13,38 +30,66 @@ | GET | `/api/agg/summary` | Сводка по флоту, список опросов upstream, `fleet_total_megabytes` / `fleet_total_connections`. Два топа (размер задаётся `top_n`): **`top_users`** — самые «прожорливые» по суммарному трафику (MiB) по всем серверам; **`top_users_by_unique_ips`** — по максимальному `active_unique_ips` среди серверов для пользователя (как в Telemt, снимок). | | GET | `/api/agg/traffic` | Трафик по каждому пользователю в разрезе серверов: `servers..total_megabytes`. | | GET | `/api/agg/unique-ips` | Уникальные IP по пользователю: на каких серверах IP есть в active/recent списках снимка. При **`geoip.enabled`** в конфиге — из City: `country_code`, `country_name`, `city_name`; при наличии ASN-БД — `asn`, `as_organization` (см. [GEOIP.md](GEOIP.md)); отключить гео для запроса: `?geo=false`. | -| GET | `/api/agg/users` | Объединённый список пользователей с `by_server` и суммарным `total_megabytes`. | +| GET | `/api/agg/users` | Объединённый список пользователей с `by_server`, суммарным `total_megabytes` и **смерженными лимитами** (см. ниже). | +| GET | `/api/agg/user/{username}` | Один пользователь в том же формате, что элементы `/api/agg/users` (без списка всех). Имя в пути: `[A-Za-z0-9_.-]+`. Ответ **`404`**, если пользователь не найден ни на одном успешном upstream. | +| GET | `/api/agg/fleet-status` | По каждому алиасу: параллельно health + system/info; в `data.servers[]` — статусы подзапросов и тела `health` / `system_info` при успехе. См. [AGGREGATE_OPENAPI.yaml](AGGREGATE_OPENAPI.yaml). | Все методы — **GET**; действует тот же whitelist, что и для остального API шлюза. +### Слияние лимитов в `users` и `user/…` + +Поля в строке пользователя (кроме счётчиков и `by_server`): + +| Поле | Политика | +| --- | --- | +| `user_ad_tag` | Первое непустое значение при обходе серверов в **лексикографическом порядке алиаса**. | +| `expiration_rfc3339` | **Самая ранняя** дата среди заданных на серверах (по разбору RFC3339 / RFC3339Nano). | +| `max_tcp_conns`, `data_quota_bytes`, `max_unique_ips` | **Минимум** среди заданных на серверах (самый строгий лимит). | + +Ссылки `links` при `include_links=true` по-прежнему берутся из **первой успешной** записи по пользователю (как раньше). + ## Query-параметры | Параметр | Где | Значение | | --- | --- | --- | | `aliases` | все | Список алиасов через запятую (например `gt1,gt2`). Если не задан — см. `aggregate.include_aliases` в YAML или все серверы из `servers`. | | `top_n` | `summary` | Размер топа пользователей (по умолчанию `10`, максимум `1000`). | -| `include_links` | `users` | `true` — добавить сгенерированные `tg://proxy` ссылки (берётся первая успешная запись по пользователю). | +| `include_links` | `users`, `user/…` | `true` — добавить сгенерированные `tg://proxy` ссылки (берётся первая успешная запись по пользователю). | | `min_total_megabytes` | `users` | Порог суммарного трафика пользователя в MiB (строго больше 0). | | `min_total_octets` | `users` | Устаревший вариант порога в октетах (если задан `min_total_megabytes`, он приоритетнее). | ## Конфигурация (опционально) ```yaml +# SPA на другом origin: список разрешённых Origin или "*" (без учётных данных cookie к шлюзу). +cors_allowed_origins: + - "http://localhost:5173" + aggregate: include_aliases: - gt1 - gt2 + # Кэш только для успешных (HTTP 200) ответов /api/agg/*, ключ = путь + query. 0 = выкл. Макс. 60000 мс. + cache_ttl_ms: 2000 ``` -Если блок отсутствует или `include_aliases` пуст, по умолчанию участвуют **все** записи `servers`. +Если блок `aggregate` отсутствует или `include_aliases` пуст, по умолчанию участвуют **все** записи `servers`. Имя алиаса **`agg`** в `servers` запрещено (зарезервировано под префикс `/api/agg/`). +### CORS + +Если задан непустой **`cors_allowed_origins`**, шлюз для подходящего заголовка **`Origin`** добавляет заголовки CORS и отвечает на **`OPTIONS`** кодом **204** без тела (preflight). Совпадение: точное равенство строки origin или `"*"`. Whitelist IP по-прежнему применяется **до** обработки запроса. + ## Ограничения - **Один и тот же `username` на разных серверах** может соответствовать разным учётным записям; агрегатор сопоставляет строки по имени — учитывайте при интерпретации сумм. - У Telemt в `UserInfo` **нет** поля «IP последний раз подключался к серверу X». В `unique-ips` поле `primary_server` заполняется **только** если ровно один сервер видит IP в `active_unique_ips_list` на момент запроса; иначе `primary_server` отсутствует или несколько серверов в списках — это снимок, не история. +## Машиночитаемый контракт + +Черновик схемы OpenAPI 3 для `/api/agg/*`: **[AGGREGATE_OPENAPI.yaml](AGGREGATE_OPENAPI.yaml)** (удобно для генерации типов на фронтенде). + ## Примеры ```bash @@ -52,4 +97,6 @@ curl -sS "http://127.0.0.1:8080/api/agg/summary" curl -sS "http://127.0.0.1:8080/api/agg/traffic?aliases=gt1,gt2" curl -sS "http://127.0.0.1:8080/api/agg/unique-ips" curl -sS "http://127.0.0.1:8080/api/agg/users?include_links=false&min_total_megabytes=1" +curl -sS "http://127.0.0.1:8080/api/agg/fleet-status" +curl -sS "http://127.0.0.1:8080/api/agg/user/myuser?aliases=gt1" ``` diff --git a/docs/AGGREGATE_OPENAPI.yaml b/docs/AGGREGATE_OPENAPI.yaml new file mode 100644 index 0000000..bc0db75 --- /dev/null +++ b/docs/AGGREGATE_OPENAPI.yaml @@ -0,0 +1,297 @@ +openapi: 3.0.3 +info: + title: Telemt API gateway — aggregate API + description: > + Эндпоинты под префиксом /api/agg на шлюзе telemt-api. + Базовый URL задаётся listen шлюза (например http://127.0.0.1:8080). + version: 1.0.0 + +paths: + /api/agg/summary: + get: + summary: Сводка флота и топы пользователей + parameters: + - $ref: '#/components/parameters/aliases' + - name: top_n + in: query + schema: { type: integer, minimum: 1, maximum: 1000, default: 10 } + responses: + '200': + description: OK + content: + application/json: + schema: { $ref: '#/components/schemas/AggEnvelopeSummary' } + + /api/agg/traffic: + get: + summary: Трафик по пользователям и серверам + parameters: + - $ref: '#/components/parameters/aliases' + responses: + '200': + description: OK + content: + application/json: + schema: { $ref: '#/components/schemas/AggEnvelopeTrafficRows' } + + /api/agg/unique-ips: + get: + summary: Уникальные IP с привязкой к серверам (и опционально GeoIP) + parameters: + - $ref: '#/components/parameters/aliases' + - name: geo + in: query + description: 'false — не обогащать GeoIP' + schema: { type: string, enum: ['false', 'true'] } + responses: + '200': + description: OK + content: + application/json: + schema: { $ref: '#/components/schemas/AggEnvelopeUniqueIPs' } + + /api/agg/users: + get: + summary: Список пользователей с merge по серверам + parameters: + - $ref: '#/components/parameters/aliases' + - name: include_links + in: query + schema: { type: string, enum: ['true', 'false'] } + - name: min_total_megabytes + in: query + schema: { type: number, format: float } + - name: min_total_octets + in: query + schema: { type: string } + responses: + '200': + description: OK + content: + application/json: + schema: { $ref: '#/components/schemas/AggEnvelopeUsersRows' } + + /api/agg/user/{username}: + get: + summary: Один пользователь (тот же объект, что в users[]) + parameters: + - name: username + in: path + required: true + schema: + type: string + pattern: '^[A-Za-z0-9_.-]+$' + - $ref: '#/components/parameters/aliases' + - name: include_links + in: query + schema: { type: string, enum: ['true', 'false'] } + responses: + '200': + description: OK + content: + application/json: + schema: { $ref: '#/components/schemas/AggEnvelopeUsersRow' } + '404': + description: Пользователь не найден ни на одном upstream + + /api/agg/fleet-status: + get: + summary: Health + system/info по всем выбранным серверам + parameters: + - $ref: '#/components/parameters/aliases' + responses: + '200': + description: OK + content: + application/json: + schema: { $ref: '#/components/schemas/AggEnvelopeFleetStatus' } + +components: + parameters: + aliases: + name: aliases + in: query + description: Список алиасов через запятую + schema: { type: string } + + schemas: + AggSuccessBase: + type: object + required: [ok, data, generated_at] + properties: + ok: { type: boolean, enum: [true] } + generated_at: { type: string, format: date-time } + partial: { type: boolean, description: true если часть upstream недоступна } + + AggEnvelopeSummary: + allOf: + - $ref: '#/components/schemas/AggSuccessBase' + - type: object + properties: + data: + type: object + description: SummaryData (см. реализацию / доку AGGREGATE.md) + + AggEnvelopeTrafficRows: + allOf: + - $ref: '#/components/schemas/AggSuccessBase' + - type: object + properties: + data: + type: array + items: { $ref: '#/components/schemas/TrafficRow' } + + AggEnvelopeUniqueIPs: + allOf: + - $ref: '#/components/schemas/AggSuccessBase' + - type: object + properties: + data: + type: array + items: { $ref: '#/components/schemas/UniqueIPsRow' } + + AggEnvelopeUsersRows: + allOf: + - $ref: '#/components/schemas/AggSuccessBase' + - type: object + properties: + data: + type: array + items: { $ref: '#/components/schemas/UsersRow' } + + AggEnvelopeUsersRow: + allOf: + - $ref: '#/components/schemas/AggSuccessBase' + - type: object + properties: + data: { $ref: '#/components/schemas/UsersRow' } + + AggEnvelopeFleetStatus: + allOf: + - $ref: '#/components/schemas/AggSuccessBase' + - type: object + properties: + data: { $ref: '#/components/schemas/FleetStatusData' } + + TrafficRow: + type: object + properties: + username: { type: string } + servers: + type: object + additionalProperties: + $ref: '#/components/schemas/TrafficServerStats' + + TrafficServerStats: + type: object + properties: + total_megabytes: { type: number, format: float } + current_connections: { type: integer, format: int64 } + revision: { type: string } + + UniqueIPsRow: + type: object + properties: + username: { type: string } + ips: + type: array + items: { $ref: '#/components/schemas/IPAssignments' } + + IPAssignments: + type: object + properties: + ip: { type: string } + active_on_servers: + type: array + items: { type: string } + recent_on_servers: + type: array + items: { type: string } + primary_server: { type: string, nullable: true } + country_code: { type: string, nullable: true } + country_name: { type: string, nullable: true } + city_name: { type: string, nullable: true } + asn: { type: integer, format: int64, nullable: true } + as_organization: { type: string, nullable: true } + + UserLinks: + type: object + properties: + classic: + type: array + items: { type: string } + secure: + type: array + items: { type: string } + tls: + type: array + items: { type: string } + + UsersRow: + type: object + properties: + username: { type: string } + total_megabytes: { type: number, format: float } + by_server: + type: object + additionalProperties: + $ref: '#/components/schemas/TrafficServerStats' + links: { $ref: '#/components/schemas/UserLinks' } + active_unique_ips: { type: integer, format: int64 } + recent_unique_ips: { type: integer, format: int64 } + user_ad_tag: { type: string, nullable: true } + max_tcp_conns: { type: integer, format: int64, nullable: true } + expiration_rfc3339: { type: string, nullable: true } + data_quota_bytes: { type: integer, format: int64, nullable: true } + max_unique_ips: { type: integer, format: int64, nullable: true } + + HealthData: + type: object + properties: + status: { type: string } + read_only: { type: boolean } + + SystemInfoData: + type: object + properties: + version: { type: string } + target_arch: { type: string } + target_os: { type: string } + build_profile: { type: string } + git_commit: { type: string, nullable: true } + build_time_utc: { type: string, nullable: true } + rustc_version: { type: string, nullable: true } + process_started_at_epoch_secs: { type: integer, format: int64 } + uptime_seconds: { type: number, format: float } + config_path: { type: string } + config_hash: { type: string } + config_reload_count: { type: integer, format: int64 } + last_config_reload_epoch_secs: { type: integer, format: int64, nullable: true } + + FleetServerStatus: + type: object + properties: + alias: { type: string } + ok: { type: boolean } + health_ok: { type: boolean } + health_http_status: { type: integer } + health_latency_ms: { type: integer, format: int64 } + health_error: { type: string } + health_revision: { type: string } + health: { $ref: '#/components/schemas/HealthData' } + system_info_ok: { type: boolean } + system_info_http_status: { type: integer } + system_info_latency_ms: { type: integer, format: int64 } + system_info_error: { type: string } + system_info_revision: { type: string } + system_info: { $ref: '#/components/schemas/SystemInfoData' } + + FleetStatusData: + type: object + properties: + servers: + type: array + items: { $ref: '#/components/schemas/FleetServerStatus' } + servers_total: { type: integer } + servers_all_ok: { type: integer } + servers_failed: { type: integer } diff --git a/internal/aggregate/fetch.go b/internal/aggregate/fetch.go index 4a53185..22a1675 100644 --- a/internal/aggregate/fetch.go +++ b/internal/aggregate/fetch.go @@ -15,105 +15,89 @@ import ( const statsUsersPath = "stats/users" -// FetchStatsUsers calls GET {base}{path_prefix}/stats/users for each alias. +// UpstreamCallMeta describes one upstream Telemt GET outcome (before typed data). +type UpstreamCallMeta struct { + OK bool `json:"ok"` + HTTPStatus int `json:"http_status,omitempty"` + LatencyMs int64 `json:"latency_ms,omitempty"` + Error string `json:"error,omitempty"` + Revision string `json:"revision,omitempty"` +} + +type telemtEnvelope[T any] struct { + OK bool `json:"ok"` + Data T `json:"data"` + Revision string `json:"revision"` +} + +// FetchTelemtGET performs GET {base}{path_prefix}/{relPath} and decodes Telemt success envelope into T. +// On upstream/network errors, returns zero T and meta with OK=false; meta.Error is set. +func FetchTelemtGET[T any](ctx context.Context, client *http.Client, parsed *config.Parsed, alias string, relPath string) (out T, meta UpstreamCallMeta) { + var zero T + srv := parsed.ByAlias[alias] + if srv == nil { + return zero, UpstreamCallMeta{OK: false, Error: "unknown alias"} + } + u, err := url.Parse(srv.BaseURL) + if err != nil { + return zero, UpstreamCallMeta{OK: false, Error: err.Error()} + } + target := joinPathPrefix(u, srv.PathPrefix, relPath) + start := time.Now() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), nil) + if err != nil { + return zero, UpstreamCallMeta{OK: false, LatencyMs: time.Since(start).Milliseconds(), Error: err.Error()} + } + if auth := parsed.AuthByAlias[alias]; auth != "" { + req.Header.Set("Authorization", auth) + } + resp, err := client.Do(req) + latency := time.Since(start).Milliseconds() + if err != nil { + return zero, UpstreamCallMeta{OK: false, LatencyMs: latency, Error: err.Error()} + } + body, readErr := io.ReadAll(io.LimitReader(resp.Body, 64<<20)) + _ = resp.Body.Close() + if readErr != nil { + return zero, UpstreamCallMeta{OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency, Error: readErr.Error()} + } + if resp.StatusCode != http.StatusOK { + return zero, UpstreamCallMeta{ + OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency, + Error: fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(body), 200)), + } + } + var env telemtEnvelope[json.RawMessage] + if err := json.Unmarshal(body, &env); err != nil { + return zero, UpstreamCallMeta{OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency, Error: "invalid json: " + err.Error()} + } + if !env.OK { + return zero, UpstreamCallMeta{OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency, Error: "upstream ok=false"} + } + var data T + if err := json.Unmarshal(env.Data, &data); err != nil { + return zero, UpstreamCallMeta{OK: false, HTTPStatus: resp.StatusCode, LatencyMs: latency, Error: "decode data: " + err.Error()} + } + return data, UpstreamCallMeta{OK: true, HTTPStatus: resp.StatusCode, LatencyMs: latency, Revision: env.Revision} +} + +// FetchStatsUsers calls GET stats/users for each alias using FetchTelemtGET. func FetchStatsUsers(ctx context.Context, client *http.Client, parsed *config.Parsed, aliases []string) []ServerFetchResult { out := make([]ServerFetchResult, 0, len(aliases)) for _, alias := range aliases { - srv := parsed.ByAlias[alias] - if srv == nil { - out = append(out, ServerFetchResult{ - Alias: alias, - OK: false, - Error: "unknown alias", - }) - continue - } - u, err := url.Parse(srv.BaseURL) - if err != nil { - out = append(out, ServerFetchResult{ - Alias: alias, - OK: false, - Error: err.Error(), - }) - continue - } - target := joinPathPrefix(u, srv.PathPrefix, statsUsersPath) - start := time.Now() - req, err := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), nil) - if err != nil { - out = append(out, ServerFetchResult{ - Alias: alias, - OK: false, - LatencyMs: time.Since(start).Milliseconds(), - Error: err.Error(), - }) - continue - } - if auth := parsed.AuthByAlias[alias]; auth != "" { - req.Header.Set("Authorization", auth) - } - resp, err := client.Do(req) - latency := time.Since(start).Milliseconds() - if err != nil { - out = append(out, ServerFetchResult{ - Alias: alias, - OK: false, - LatencyMs: latency, - Error: err.Error(), - }) - continue - } - body, readErr := io.ReadAll(io.LimitReader(resp.Body, 64<<20)) - _ = resp.Body.Close() - if readErr != nil { - out = append(out, ServerFetchResult{ - Alias: alias, - OK: false, - HTTPStatus: resp.StatusCode, - LatencyMs: latency, - Error: readErr.Error(), - }) - continue - } - if resp.StatusCode != http.StatusOK { - out = append(out, ServerFetchResult{ - Alias: alias, - OK: false, - HTTPStatus: resp.StatusCode, - LatencyMs: latency, - Error: fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(body), 200)), - }) - continue - } - var env statsUsersEnvelope - if err := json.Unmarshal(body, &env); err != nil { - out = append(out, ServerFetchResult{ - Alias: alias, - OK: false, - HTTPStatus: resp.StatusCode, - LatencyMs: latency, - Error: "invalid json: " + err.Error(), - }) - continue - } - if !env.OK { - out = append(out, ServerFetchResult{ - Alias: alias, - OK: false, - HTTPStatus: resp.StatusCode, - LatencyMs: latency, - Error: "upstream ok=false", - }) - continue - } - out = append(out, ServerFetchResult{ + users, meta := FetchTelemtGET[[]UserInfo](ctx, client, parsed, alias, statsUsersPath) + fr := ServerFetchResult{ Alias: alias, - OK: true, - HTTPStatus: resp.StatusCode, - LatencyMs: latency, - Revision: env.Revision, - Users: env.Data, - }) + OK: meta.OK, + HTTPStatus: meta.HTTPStatus, + LatencyMs: meta.LatencyMs, + Error: meta.Error, + Revision: meta.Revision, + } + if meta.OK { + fr.Users = users + } + out = append(out, fr) } return out } diff --git a/internal/aggregate/fleet.go b/internal/aggregate/fleet.go new file mode 100644 index 0000000..1865b12 --- /dev/null +++ b/internal/aggregate/fleet.go @@ -0,0 +1,67 @@ +package aggregate + +import ( + "context" + "net/http" + "sort" + "sync" + + "github.com/telemt/telemt-api/internal/config" +) + +// FetchFleetStatus probes GET health and GET system/info for each alias in parallel. +func FetchFleetStatus(ctx context.Context, client *http.Client, parsed *config.Parsed, aliases []string) FleetStatusData { + if len(aliases) == 0 { + return FleetStatusData{} + } + rows := make([]FleetServerStatus, len(aliases)) + var wg sync.WaitGroup + for i, alias := range aliases { + i, alias := i, alias + wg.Add(1) + go func() { + defer wg.Done() + rows[i] = probeFleetServer(ctx, client, parsed, alias) + }() + } + wg.Wait() + sort.Slice(rows, func(i, j int) bool { return rows[i].Alias < rows[j].Alias }) + allOK, failed := 0, 0 + for _, r := range rows { + if r.OK { + allOK++ + } else { + failed++ + } + } + return FleetStatusData{ + Servers: rows, + ServersTotal: len(rows), + ServersAllOK: allOK, + ServersFailed: failed, + } +} + +func probeFleetServer(ctx context.Context, client *http.Client, parsed *config.Parsed, alias string) FleetServerStatus { + h, hm := FetchTelemtGET[HealthData](ctx, client, parsed, alias, "health") + s, sm := FetchTelemtGET[SystemInfoData](ctx, client, parsed, alias, "system/info") + row := FleetServerStatus{Alias: alias} + row.HealthOK = hm.OK + row.HealthHTTPStatus = hm.HTTPStatus + row.HealthLatencyMs = hm.LatencyMs + row.HealthError = hm.Error + row.HealthRevision = hm.Revision + if hm.OK { + row.Health = &h + } + row.SystemInfoOK = sm.OK + row.SystemInfoHTTPStatus = sm.HTTPStatus + row.SystemInfoLatencyMs = sm.LatencyMs + row.SystemInfoError = sm.Error + row.SystemInfoRevision = sm.Revision + if sm.OK { + row.SystemInfo = &s + } + row.OK = hm.OK && sm.OK + return row +} diff --git a/internal/aggregate/handlers.go b/internal/aggregate/handlers.go index 7267e97..231e8a6 100644 --- a/internal/aggregate/handlers.go +++ b/internal/aggregate/handlers.go @@ -4,8 +4,11 @@ import ( "context" "encoding/json" "net/http" + "net/http/httptest" + "regexp" "strconv" "strings" + "sync" "time" "github.com/telemt/telemt-api/internal/config" @@ -14,17 +17,34 @@ import ( const pathPrefix = "/api/agg" +var aggUsernameRe = regexp.MustCompile(`^[A-Za-z0-9_.-]+$`) + +type cacheEntry struct { + body []byte + expires time.Time +} + // Handler serves GET /api/agg/* aggregate endpoints. type Handler struct { - Parsed *config.Parsed - Client *http.Client - Geo *geoip.Service + Parsed *config.Parsed + Client *http.Client + Geo *geoip.Service + CacheTTL time.Duration + + cacheMu sync.Mutex + cache map[string]cacheEntry } // NewHandler builds an aggregate handler; client must use a non-nil Transport (e.g. gateway shared transport). -// Geo may be nil (no GeoLite2 lookups). -func NewHandler(p *config.Parsed, client *http.Client, geo *geoip.Service) *Handler { - return &Handler{Parsed: p, Client: client, Geo: geo} +// Geo may be nil (no GeoLite2 lookups). cacheTTL 0 disables response caching. +func NewHandler(p *config.Parsed, client *http.Client, geo *geoip.Service, cacheTTL time.Duration) *Handler { + return &Handler{ + Parsed: p, + Client: client, + Geo: geo, + CacheTTL: cacheTTL, + cache: make(map[string]cacheEntry), + } } func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -36,19 +56,67 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } sub := strings.TrimPrefix(r.URL.Path, pathPrefix) sub = strings.TrimPrefix(sub, "/") - switch sub { - case "summary": + + if h.CacheTTL > 0 { + key := r.URL.Path + "\x00" + r.URL.RawQuery + now := time.Now() + h.cacheMu.Lock() + ent, hit := h.cache[key] + if hit && now.Before(ent.expires) { + body := ent.body + h.cacheMu.Unlock() + w.Header().Set("Content-Type", "application/json; charset=utf-8") + _, _ = w.Write(body) + return + } + h.cacheMu.Unlock() + + rec := httptest.NewRecorder() + h.dispatch(rec, r, sub) + body := rec.Body.Bytes() + if rec.Code == http.StatusOK { + h.cacheMu.Lock() + h.cache[key] = cacheEntry{body: append([]byte(nil), body...), expires: now.Add(h.CacheTTL)} + h.cacheMu.Unlock() + } + copyRecorderToResponse(rec, w) + return + } + + h.dispatch(w, r, sub) +} + +func copyRecorderToResponse(rec *httptest.ResponseRecorder, w http.ResponseWriter) { + for k, vv := range rec.Header() { + for _, v := range vv { + w.Header().Add(k, v) + } + } + w.WriteHeader(rec.Code) + _, _ = w.Write(rec.Body.Bytes()) +} + +func (h *Handler) dispatch(w http.ResponseWriter, r *http.Request, sub string) { + switch { + case sub == "summary": h.handleSummary(w, r) - case "traffic": + case sub == "traffic": h.handleTraffic(w, r) - case "unique-ips": + case sub == "unique-ips": h.handleUniqueIPs(w, r) - case "users": + case sub == "users": h.handleUsers(w, r) + case sub == "fleet-status": + h.handleFleetStatus(w, r) + case strings.HasPrefix(sub, "user/"): + username := strings.TrimPrefix(sub, "user/") + if username == "" { + writeNotFound(w, "missing username") + return + } + h.handleUserOne(w, r, username) default: - w.Header().Set("Content-Type", "application/json; charset=utf-8") - w.WriteHeader(http.StatusNotFound) - _ = json.NewEncoder(w).Encode(errEnvelope("not_found", "unknown aggregate path")) + writeNotFound(w, "unknown aggregate path") } } @@ -115,6 +183,15 @@ type resolveError struct { func (e *resolveError) Error() string { return e.msg } +func anyUpstreamFailed(results []ServerFetchResult) bool { + for _, fr := range results { + if !fr.OK { + return true + } + } + return false +} + func (h *Handler) handleSummary(w http.ResponseWriter, r *http.Request) { aliases, err := h.resolveAliases(r) if err != nil { @@ -131,7 +208,8 @@ func (h *Handler) handleSummary(w http.ResponseWriter, r *http.Request) { defer cancel() results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases) data := BuildSummary(results, topN) - writeOK(w, data) + partial := anyUpstreamFailed(results) + writeAggOK(w, partial, data) } func (h *Handler) handleTraffic(w http.ResponseWriter, r *http.Request) { @@ -144,7 +222,8 @@ func (h *Handler) handleTraffic(w http.ResponseWriter, r *http.Request) { defer cancel() results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases) data := BuildTraffic(results) - writeOK(w, data) + partial := anyUpstreamFailed(results) + writeAggOK(w, partial, data) } func (h *Handler) handleUniqueIPs(w http.ResponseWriter, r *http.Request) { @@ -160,7 +239,8 @@ func (h *Handler) handleUniqueIPs(w http.ResponseWriter, r *http.Request) { if h.Geo != nil && !strings.EqualFold(r.URL.Query().Get("geo"), "false") { EnrichUniqueIPsGeo(data, h.Geo) } - writeOK(w, data) + partial := anyUpstreamFailed(results) + writeAggOK(w, partial, data) } func (h *Handler) handleUsers(w http.ResponseWriter, r *http.Request) { @@ -184,7 +264,44 @@ func (h *Handler) handleUsers(w http.ResponseWriter, r *http.Request) { defer cancel() results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases) data := BuildUsers(results, includeLinks, minOct) - writeOK(w, data) + partial := anyUpstreamFailed(results) + writeAggOK(w, partial, data) +} + +func (h *Handler) handleFleetStatus(w http.ResponseWriter, r *http.Request) { + aliases, err := h.resolveAliases(r) + if err != nil { + writeBadRequest(w, err) + return + } + ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second) + defer cancel() + data := FetchFleetStatus(ctx, h.Client, h.Parsed, aliases) + partial := data.ServersFailed > 0 + writeAggOK(w, partial, data) +} + +func (h *Handler) handleUserOne(w http.ResponseWriter, r *http.Request, username string) { + if !aggUsernameRe.MatchString(username) { + writeBadRequestString(w, "invalid username") + return + } + aliases, err := h.resolveAliases(r) + if err != nil { + writeBadRequest(w, err) + return + } + includeLinks := strings.EqualFold(r.URL.Query().Get("include_links"), "true") + ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second) + defer cancel() + results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases) + row := BuildSingleUser(results, username, includeLinks) + if row == nil { + writeNotFound(w, "user not found on any upstream") + return + } + partial := anyUpstreamFailed(results) + writeAggOK(w, partial, row) } func writeBadRequest(w http.ResponseWriter, err error) { @@ -193,7 +310,27 @@ func writeBadRequest(w http.ResponseWriter, err error) { _ = json.NewEncoder(w).Encode(errEnvelope("bad_request", err.Error())) } -func writeOK(w http.ResponseWriter, data any) { +func writeBadRequestString(w http.ResponseWriter, msg string) { w.Header().Set("Content-Type", "application/json; charset=utf-8") - _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "data": data}) + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(errEnvelope("bad_request", msg)) +} + +func writeNotFound(w http.ResponseWriter, msg string) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(http.StatusNotFound) + _ = json.NewEncoder(w).Encode(errEnvelope("not_found", msg)) +} + +func writeAggOK(w http.ResponseWriter, partial bool, data any) { + env := map[string]any{ + "ok": true, + "data": data, + "generated_at": time.Now().UTC().Format(time.RFC3339Nano), + } + if partial { + env["partial"] = true + } + w.Header().Set("Content-Type", "application/json; charset=utf-8") + _ = json.NewEncoder(w).Encode(env) } diff --git a/internal/aggregate/handlers_test.go b/internal/aggregate/handlers_test.go index 27743f2..fcab56c 100644 --- a/internal/aggregate/handlers_test.go +++ b/internal/aggregate/handlers_test.go @@ -35,7 +35,7 @@ func TestHandlerResolveAndFetch(t *testing.T) { if err != nil { t.Fatal(err) } - h := NewHandler(parsed, up.Client(), nil) + h := NewHandler(parsed, up.Client(), nil, 0) req := httptest.NewRequest(http.MethodGet, "/api/agg/summary?aliases=test", nil) rec := httptest.NewRecorder() h.ServeHTTP(rec, req) @@ -54,6 +54,135 @@ func TestHandlerResolveAndFetch(t *testing.T) { } } +func TestHandlerFleetStatus(t *testing.T) { + up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/health": + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": true, "data": map[string]any{"status": "ok", "read_only": false}, "revision": "rh", + }) + case "/v1/system/info": + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": true, + "data": map[string]any{ + "version": "1.0.0", "target_arch": "amd64", "target_os": "linux", "build_profile": "release", + "process_started_at_epoch_secs": 1, "uptime_seconds": 10.0, "config_path": "/x.toml", + "config_hash": "abc", "config_reload_count": uint64(0), + }, + "revision": "rs", + }) + default: + http.NotFound(w, r) + } + })) + defer up.Close() + + cfg := &config.Config{ + Servers: []config.Server{ + {Alias: "test", BaseURL: up.URL, PathPrefix: "/v1"}, + }, + } + if err := cfg.Validate(); err != nil { + t.Fatal(err) + } + parsed, err := cfg.Parse() + if err != nil { + t.Fatal(err) + } + h := NewHandler(parsed, up.Client(), nil, 0) + req := httptest.NewRequest(http.MethodGet, "/api/agg/fleet-status?aliases=test", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status %d body %s", rec.Code, rec.Body.String()) + } + var env struct { + OK bool `json:"ok"` + Data FleetStatusData `json:"data"` + GeneratedAt string `json:"generated_at"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil { + t.Fatal(err) + } + if !env.OK || env.GeneratedAt == "" || len(env.Data.Servers) != 1 { + t.Fatalf("envelope: %+v", env) + } + s := env.Data.Servers[0] + if s.Alias != "test" || !s.OK || s.Health == nil || s.SystemInfo == nil { + t.Fatalf("server row: %+v", s) + } +} + +func TestHandlerUserOne(t *testing.T) { + up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/stats/users" { + http.NotFound(w, r) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": true, + "data": []map[string]any{{ + "username": "u1", "total_octets": 1048576, "current_connections": 2, + "active_unique_ips": 1, "recent_unique_ips": 1, + "max_tcp_conns": 10, + "data_quota_bytes": 1000, + }}, + "revision": "abc", + }) + })) + defer up.Close() + + cfg := &config.Config{ + Servers: []config.Server{ + {Alias: "test", BaseURL: up.URL, PathPrefix: "/v1"}, + }, + } + if err := cfg.Validate(); err != nil { + t.Fatal(err) + } + parsed, err := cfg.Parse() + if err != nil { + t.Fatal(err) + } + h := NewHandler(parsed, up.Client(), nil, 0) + req := httptest.NewRequest(http.MethodGet, "/api/agg/user/u1?aliases=test", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status %d body %s", rec.Code, rec.Body.String()) + } + var env struct { + OK bool `json:"ok"` + Data UsersRow `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil { + t.Fatal(err) + } + if env.Data.Username != "u1" || env.Data.MaxTCPConns == nil || *env.Data.MaxTCPConns != 10 { + t.Fatalf("row: %+v", env.Data) + } +} + +func TestHandlerUserOneInvalidName(t *testing.T) { + cfg := &config.Config{ + Servers: []config.Server{{Alias: "x", BaseURL: "http://127.0.0.1:1", PathPrefix: "/v1"}}, + } + if err := cfg.Validate(); err != nil { + t.Fatal(err) + } + parsed, err := cfg.Parse() + if err != nil { + t.Fatal(err) + } + h := NewHandler(parsed, http.DefaultClient, nil, 0) + req := httptest.NewRequest(http.MethodGet, "/api/agg/user/!!!", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("got %d", rec.Code) + } +} + func TestHandlerMethodNotAllowed(t *testing.T) { cfg := &config.Config{ Servers: []config.Server{{Alias: "x", BaseURL: "http://127.0.0.1:1", PathPrefix: "/v1"}}, @@ -65,7 +194,7 @@ func TestHandlerMethodNotAllowed(t *testing.T) { if err != nil { t.Fatal(err) } - h := NewHandler(parsed, http.DefaultClient, nil) + h := NewHandler(parsed, http.DefaultClient, nil, 0) req := httptest.NewRequest(http.MethodPost, "/api/agg/summary", nil) rec := httptest.NewRecorder() h.ServeHTTP(rec, req) diff --git a/internal/aggregate/merge.go b/internal/aggregate/merge.go index 159be18..c7e966c 100644 --- a/internal/aggregate/merge.go +++ b/internal/aggregate/merge.go @@ -3,6 +3,7 @@ package aggregate import ( "sort" "strings" + "time" ) // BuildTraffic builds traffic matrix from successful fetches only. @@ -140,6 +141,11 @@ func sortedKeys(m map[string]struct{}) []string { return s } +type userOnServer struct { + alias string + u UserInfo +} + // BuildUsers merged rows with totals and optional links from first successful server per user. func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets uint64) []UsersRow { type acc struct { @@ -150,6 +156,7 @@ func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets u rec uint64 } m := map[string]*acc{} + perUserServers := map[string][]userOnServer{} for _, fr := range results { if !fr.OK { @@ -176,6 +183,7 @@ func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets u if includeLinks && u.Links != nil && a.links == nil { a.links = u.Links } + perUserServers[u.Username] = append(perUserServers[u.Username], userOnServer{alias: fr.Alias, u: u}) } } @@ -190,18 +198,99 @@ func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets u rows := make([]UsersRow, 0, len(names)) for _, name := range names { a := m[name] + sort.Slice(perUserServers[name], func(i, j int) bool { + return perUserServers[name][i].alias < perUserServers[name][j].alias + }) + ad, mt, ex, dq, mu := mergeUserLimitFields(perUserServers[name]) rows = append(rows, UsersRow{ - Username: name, - TotalMegabytes: octetsToMegabytes(a.total), - ByServer: a.byServer, - Links: a.links, - ActiveUniqueIPs: a.act, - RecentUniqueIPs: a.rec, + Username: name, + TotalMegabytes: octetsToMegabytes(a.total), + ByServer: a.byServer, + Links: a.links, + ActiveUniqueIPs: a.act, + RecentUniqueIPs: a.rec, + UserAdTag: ad, + MaxTCPConns: mt, + ExpirationRFC3339: ex, + DataQuotaBytes: dq, + MaxUniqueIPs: mu, }) } return rows } +// BuildSingleUser returns one merged user row or nil if the user is absent on all successful upstreams. +func BuildSingleUser(results []ServerFetchResult, username string, includeLinks bool) *UsersRow { + rows := BuildUsers(results, includeLinks, 0) + for i := range rows { + if rows[i].Username == username { + return &rows[i] + } + } + return nil +} + +func mergeUserLimitFields(rows []userOnServer) (userAdTag *string, maxTCP *uint64, exp *string, quota *uint64, maxUip *uint64) { + for _, r := range rows { + if r.u.UserAdTag != nil && userAdTag == nil { + v := *r.u.UserAdTag + userAdTag = &v + } + } + var earliest time.Time + var earliestStr string + haveEarliest := false + for _, r := range rows { + if r.u.ExpirationRFC3339 == nil { + continue + } + s := *r.u.ExpirationRFC3339 + t, err := time.Parse(time.RFC3339Nano, s) + if err != nil { + t, err = time.Parse(time.RFC3339, s) + } + if err != nil { + continue + } + if !haveEarliest || t.Before(earliest) { + earliest = t + earliestStr = s + haveEarliest = true + } + } + if haveEarliest { + exp = &earliestStr + } + for _, r := range rows { + if r.u.MaxTCPConns != nil { + v := *r.u.MaxTCPConns + if maxTCP == nil || v < *maxTCP { + maxTCP = new(uint64) + *maxTCP = v + } + } + } + for _, r := range rows { + if r.u.DataQuotaBytes != nil { + v := *r.u.DataQuotaBytes + if quota == nil || v < *quota { + quota = new(uint64) + *quota = v + } + } + } + for _, r := range rows { + if r.u.MaxUniqueIPs != nil { + v := *r.u.MaxUniqueIPs + if maxUip == nil || v < *maxUip { + maxUip = new(uint64) + *maxUip = v + } + } + } + return userAdTag, maxTCP, exp, quota, maxUip +} + // BuildSummary computes fleet totals and top users. func BuildSummary(results []ServerFetchResult, topN int) SummaryData { if topN <= 0 { diff --git a/internal/aggregate/merge_test.go b/internal/aggregate/merge_test.go index c02a6b2..0413582 100644 --- a/internal/aggregate/merge_test.go +++ b/internal/aggregate/merge_test.go @@ -119,6 +119,55 @@ func TestBuildSummary(t *testing.T) { } } +func TestBuildUsersMergeLimits(t *testing.T) { + tag1 := "aa" + tag2 := "bb" + expLater := "2030-01-02T00:00:00Z" + expEarlier := "2020-01-02T00:00:00Z" + m20 := uint64(20) + m10 := uint64(10) + q500 := uint64(500) + q100 := uint64(100) + results := []ServerFetchResult{ + { + Alias: "b", OK: true, + Users: []UserInfo{{ + Username: "u", TotalOctets: 1, CurrentConnections: 0, + UserAdTag: &tag2, MaxTCPConns: &m20, ExpirationRFC3339: &expLater, + DataQuotaBytes: &q500, MaxUniqueIPs: &m10, + }}, + }, + { + Alias: "a", OK: true, + Users: []UserInfo{{ + Username: "u", TotalOctets: 1, CurrentConnections: 0, + UserAdTag: &tag1, MaxTCPConns: &m10, ExpirationRFC3339: &expEarlier, + DataQuotaBytes: &q100, MaxUniqueIPs: &m20, + }}, + }, + } + rows := BuildUsers(results, false, 0) + if len(rows) != 1 { + t.Fatalf("rows: %+v", rows) + } + r := rows[0] + if r.UserAdTag == nil || *r.UserAdTag != tag1 { + t.Fatalf("ad tag want first alias order a: got %v", r.UserAdTag) + } + if r.MaxTCPConns == nil || *r.MaxTCPConns != 10 { + t.Fatalf("max tcp want min 10: %v", r.MaxTCPConns) + } + if r.ExpirationRFC3339 == nil || *r.ExpirationRFC3339 != expEarlier { + t.Fatalf("expiration want earliest: %v", r.ExpirationRFC3339) + } + if r.DataQuotaBytes == nil || *r.DataQuotaBytes != 100 { + t.Fatalf("quota want min: %v", r.DataQuotaBytes) + } + if r.MaxUniqueIPs == nil || *r.MaxUniqueIPs != 10 { + t.Fatalf("max unique ips want min: %v", r.MaxUniqueIPs) + } +} + func TestBuildUsersMinOctets(t *testing.T) { results := []ServerFetchResult{ {Alias: "a", OK: true, Users: []UserInfo{{Username: "low", TotalOctets: 5}}}, diff --git a/internal/aggregate/types.go b/internal/aggregate/types.go index cb19fd4..41cb5b6 100644 --- a/internal/aggregate/types.go +++ b/internal/aggregate/types.go @@ -24,10 +24,55 @@ type UserLinks struct { TLS []string `json:"tls,omitempty"` } -type statsUsersEnvelope struct { - OK bool `json:"ok"` - Data []UserInfo `json:"data"` - Revision string `json:"revision"` +// HealthData mirrors Telemt GET /v1/health data block. +type HealthData struct { + Status string `json:"status"` + ReadOnly bool `json:"read_only"` +} + +// SystemInfoData mirrors Telemt GET /v1/system/info (subset used by fleet-status UI). +type SystemInfoData struct { + Version string `json:"version"` + TargetArch string `json:"target_arch"` + TargetOS string `json:"target_os"` + BuildProfile string `json:"build_profile"` + GitCommit *string `json:"git_commit"` + BuildTimeUTC *string `json:"build_time_utc"` + RustcVersion *string `json:"rustc_version"` + ProcessStartedAtEpochSecs uint64 `json:"process_started_at_epoch_secs"` + UptimeSeconds float64 `json:"uptime_seconds"` + ConfigPath string `json:"config_path"` + ConfigHash string `json:"config_hash"` + ConfigReloadCount uint64 `json:"config_reload_count"` + LastConfigReloadEpochSecs *uint64 `json:"last_config_reload_epoch_secs"` +} + +// FleetServerStatus is one upstream health + system/info probe. +type FleetServerStatus struct { + Alias string `json:"alias"` + OK bool `json:"ok"` + + HealthOK bool `json:"health_ok"` + HealthHTTPStatus int `json:"health_http_status,omitempty"` + HealthLatencyMs int64 `json:"health_latency_ms,omitempty"` + HealthError string `json:"health_error,omitempty"` + HealthRevision string `json:"health_revision,omitempty"` + Health *HealthData `json:"health,omitempty"` + + SystemInfoOK bool `json:"system_info_ok"` + SystemInfoHTTPStatus int `json:"system_info_http_status,omitempty"` + SystemInfoLatencyMs int64 `json:"system_info_latency_ms,omitempty"` + SystemInfoError string `json:"system_info_error,omitempty"` + SystemInfoRevision string `json:"system_info_revision,omitempty"` + SystemInfo *SystemInfoData `json:"system_info,omitempty"` +} + +// FleetStatusData aggregates fleet-status across aliases. +type FleetStatusData struct { + Servers []FleetServerStatus `json:"servers"` + ServersTotal int `json:"servers_total"` + ServersAllOK int `json:"servers_all_ok"` + ServersFailed int `json:"servers_failed"` } // ServerFetchResult is one upstream GET /v1/stats/users outcome. @@ -77,11 +122,17 @@ type IPAssignments struct { // UsersRow merged user view with optional per-server detail and links. type UsersRow struct { Username string `json:"username"` - TotalMegabytes float64 `json:"total_megabytes"` + TotalMegabytes float64 `json:"total_megabytes"` ByServer map[string]TrafficServerStats `json:"by_server"` Links *UserLinks `json:"links,omitempty"` ActiveUniqueIPs uint64 `json:"active_unique_ips"` RecentUniqueIPs uint64 `json:"recent_unique_ips"` + // Merged limits across servers (policy: docs/AGGREGATE.md). + UserAdTag *string `json:"user_ad_tag,omitempty"` + MaxTCPConns *uint64 `json:"max_tcp_conns,omitempty"` + ExpirationRFC3339 *string `json:"expiration_rfc3339,omitempty"` + DataQuotaBytes *uint64 `json:"data_quota_bytes,omitempty"` + MaxUniqueIPs *uint64 `json:"max_unique_ips,omitempty"` } // SummaryData fleet snapshot. diff --git a/internal/config/config.go b/internal/config/config.go index 4d83614..f221232 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -15,13 +15,14 @@ var aliasRe = regexp.MustCompile(`^[a-z0-9][a-z0-9_-]*$`) // Config is the gateway YAML configuration. type Config struct { - Listen string `yaml:"listen"` - AllowAll bool `yaml:"allow_all"` - WhitelistCIDRs []string `yaml:"whitelist_cidrs"` - TrustedProxies []string `yaml:"trusted_proxies"` - Servers []Server `yaml:"servers"` - Aggregate *AggregateConfig `yaml:"aggregate"` - GeoIP *GeoIPConfig `yaml:"geoip"` + Listen string `yaml:"listen"` + AllowAll bool `yaml:"allow_all"` + WhitelistCIDRs []string `yaml:"whitelist_cidrs"` + TrustedProxies []string `yaml:"trusted_proxies"` + CorsAllowedOrigins []string `yaml:"cors_allowed_origins"` + Servers []Server `yaml:"servers"` + Aggregate *AggregateConfig `yaml:"aggregate"` + GeoIP *GeoIPConfig `yaml:"geoip"` } // GeoIPConfig enables GeoLite2 lookups for /api/agg/unique-ips (optional). @@ -38,6 +39,8 @@ type GeoIPConfig struct { type AggregateConfig struct { // IncludeAliases limits aggregation to these server aliases; empty means all servers. IncludeAliases []string `yaml:"include_aliases"` + // CacheTTLMs is in-memory cache TTL for successful GET /api/agg/* responses (milliseconds). 0 disables. + CacheTTLMs uint64 `yaml:"cache_ttl_ms"` } // Server maps a URL alias to an upstream base URL. @@ -123,6 +126,9 @@ func (c *Config) Validate() error { return fmt.Errorf("aggregate.include_aliases[%d]: unknown server alias %q", i, a) } } + if c.Aggregate.CacheTTLMs > 60000 { + return fmt.Errorf("aggregate.cache_ttl_ms must be within [0, 60000]") + } } return nil } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 5c73f23..3090cd2 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -98,3 +98,19 @@ func TestValidateAggregateIncludeAliases(t *testing.T) { t.Fatal("expected error for unknown include alias") } } + +func TestValidateAggregateCacheTTL(t *testing.T) { + c := &Config{ + Servers: []Server{ + {Alias: "a", BaseURL: "http://x:1"}, + }, + Aggregate: &AggregateConfig{CacheTTLMs: 60001}, + } + if err := c.Validate(); err == nil { + t.Fatal("expected error for cache_ttl_ms > 60000") + } + c.Aggregate.CacheTTLMs = 1000 + if err := c.Validate(); err != nil { + t.Fatal(err) + } +} diff --git a/internal/server/gateway.go b/internal/server/gateway.go index cab7919..e483902 100644 --- a/internal/server/gateway.go +++ b/internal/server/gateway.go @@ -28,6 +28,7 @@ type Gateway struct { log *slog.Logger transport *http.Transport promHandler http.Handler + corsAllowed []string } // NewGateway builds handlers and reverse proxies from parsed config. @@ -68,19 +69,68 @@ func NewGateway(p *config.Parsed, log *slog.Logger, geo *geoip.Service) (*Gatewa } g.proxies[s.Alias] = rp } - g.agg = aggregate.NewHandler(p, &http.Client{Transport: t}, geo) + var aggCacheTTL time.Duration + if p.Config.Aggregate != nil && p.Config.Aggregate.CacheTTLMs > 0 { + aggCacheTTL = time.Duration(p.Config.Aggregate.CacheTTLMs) * time.Millisecond + } + g.corsAllowed = append([]string(nil), p.Config.CorsAllowedOrigins...) + g.agg = aggregate.NewHandler(p, &http.Client{Transport: t}, geo, aggCacheTTL) return g, nil } // Handler returns the root HTTP handler with middleware. func (g *Gateway) Handler() http.Handler { var h http.Handler = http.HandlerFunc(g.serve) + h = g.withCORS(h) h = g.withWhitelist(h) h = g.withAccessLog(h) h = g.withMetrics(h) return h } +func (g *Gateway) withCORS(next http.Handler) http.Handler { + if len(g.corsAllowed) == 0 { + return next + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Add("Vary", "Origin") + origin := r.Header.Get("Origin") + ok, allowOrigin := corsMatch(g.corsAllowed, origin) + if ok { + w.Header().Set("Access-Control-Allow-Origin", allowOrigin) + w.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type, X-Request-Id") + w.Header().Set("Access-Control-Max-Age", "86400") + } + if r.Method == http.MethodOptions { + if ok { + w.WriteHeader(http.StatusNoContent) + return + } + } + next.ServeHTTP(w, r) + }) +} + +func corsMatch(allowed []string, origin string) (ok bool, allowOrigin string) { + if origin == "" { + return false, "" + } + for _, a := range allowed { + a = strings.TrimSpace(a) + if a == "" { + continue + } + if a == "*" { + return true, "*" + } + if strings.EqualFold(a, origin) { + return true, origin + } + } + return false, "" +} + func (g *Gateway) withWhitelist(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/health" {