Add CORS support and response caching to aggregate endpoints
- Introduced CORS configuration options in config.example.yaml, allowing specification of allowed origins for cross-origin requests. - Enhanced the aggregate handler to support response caching with a configurable TTL, improving performance for repeated requests. - Updated the aggregate API to return a structured response indicating whether any upstream requests failed, enhancing error handling and response clarity. - Modified documentation in AGGREGATE.md and README.md to reflect the new CORS and caching features. - Added tests to validate the new functionality in the aggregate handler.
This commit is contained in:
@@ -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/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/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` |
|
| **[docs/GEOIP.md](docs/GEOIP.md)** | GeoLite2 City (страна/город) и опционально ASN (номер AS, организация) для IP в `unique-ips` |
|
||||||
|
|
||||||
## Сборка и тесты без Docker
|
## Сборка и тесты без Docker
|
||||||
|
|||||||
+7
-1
@@ -23,11 +23,17 @@ whitelist_cidrs:
|
|||||||
# - "10.0.0.0/8"
|
# - "10.0.0.0/8"
|
||||||
trusted_proxies: []
|
trusted_proxies: []
|
||||||
|
|
||||||
# Опционально: по умолчанию /api/agg/* опрашивает все servers; можно ограничить список:
|
# SPA на другом origin (preflight OPTIONS + Access-Control-Allow-Origin):
|
||||||
|
# cors_allowed_origins:
|
||||||
|
# - "http://localhost:5173"
|
||||||
|
# # - "*"
|
||||||
|
|
||||||
|
# Опционально: по умолчанию /api/agg/* опрашивает все servers; можно ограничить список и включить кэш ответов:
|
||||||
# aggregate:
|
# aggregate:
|
||||||
# include_aliases:
|
# include_aliases:
|
||||||
# - gt1
|
# - gt1
|
||||||
# - gt2
|
# - gt2
|
||||||
|
# cache_ttl_ms: 2000
|
||||||
|
|
||||||
# Геолокация IP в /api/agg/unique-ips. См. docs/GEOIP.md (City + опционально ASN; Country-only не нужен)
|
# Геолокация IP в /api/agg/unique-ips. См. docs/GEOIP.md (City + опционально ASN; Country-only не нужен)
|
||||||
# geoip:
|
# geoip:
|
||||||
|
|||||||
+51
-4
@@ -1,11 +1,28 @@
|
|||||||
# Агрегирующие эндпоинты шлюза (`/api/agg/`)
|
# Агрегирующие эндпоинты шлюза (`/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 для удобства).
|
**Единицы трафика в агрегатах:** поля `*_megabytes` — это **двоичные мегабайты (MiB)**, 1 MiB = 1024² октетов (как у Telemt в ответе считаются октеты, шлюз делит на MiB для удобства).
|
||||||
|
|
||||||
Доступ к **одному** инстансу по-прежнему через прокси: `GET /api/{alias}/…` (например `/api/gt1/v1/stats/users`) — там по-прежнему `total_octets` как в [API.md](API.md).
|
Доступ к **одному** инстансу по-прежнему через прокси: `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/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.<alias>.total_megabytes`. |
|
| GET | `/api/agg/traffic` | Трафик по каждому пользователю в разрезе серверов: `servers.<alias>.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/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 шлюза.
|
Все методы — **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-параметры
|
## Query-параметры
|
||||||
|
|
||||||
| Параметр | Где | Значение |
|
| Параметр | Где | Значение |
|
||||||
| --- | --- | --- |
|
| --- | --- | --- |
|
||||||
| `aliases` | все | Список алиасов через запятую (например `gt1,gt2`). Если не задан — см. `aggregate.include_aliases` в YAML или все серверы из `servers`. |
|
| `aliases` | все | Список алиасов через запятую (например `gt1,gt2`). Если не задан — см. `aggregate.include_aliases` в YAML или все серверы из `servers`. |
|
||||||
| `top_n` | `summary` | Размер топа пользователей (по умолчанию `10`, максимум `1000`). |
|
| `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_megabytes` | `users` | Порог суммарного трафика пользователя в MiB (строго больше 0). |
|
||||||
| `min_total_octets` | `users` | Устаревший вариант порога в октетах (если задан `min_total_megabytes`, он приоритетнее). |
|
| `min_total_octets` | `users` | Устаревший вариант порога в октетах (если задан `min_total_megabytes`, он приоритетнее). |
|
||||||
|
|
||||||
## Конфигурация (опционально)
|
## Конфигурация (опционально)
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
|
# SPA на другом origin: список разрешённых Origin или "*" (без учётных данных cookie к шлюзу).
|
||||||
|
cors_allowed_origins:
|
||||||
|
- "http://localhost:5173"
|
||||||
|
|
||||||
aggregate:
|
aggregate:
|
||||||
include_aliases:
|
include_aliases:
|
||||||
- gt1
|
- gt1
|
||||||
- gt2
|
- gt2
|
||||||
|
# Кэш только для успешных (HTTP 200) ответов /api/agg/*, ключ = путь + query. 0 = выкл. Макс. 60000 мс.
|
||||||
|
cache_ttl_ms: 2000
|
||||||
```
|
```
|
||||||
|
|
||||||
Если блок отсутствует или `include_aliases` пуст, по умолчанию участвуют **все** записи `servers`.
|
Если блок `aggregate` отсутствует или `include_aliases` пуст, по умолчанию участвуют **все** записи `servers`.
|
||||||
|
|
||||||
Имя алиаса **`agg`** в `servers` запрещено (зарезервировано под префикс `/api/agg/`).
|
Имя алиаса **`agg`** в `servers` запрещено (зарезервировано под префикс `/api/agg/`).
|
||||||
|
|
||||||
|
### CORS
|
||||||
|
|
||||||
|
Если задан непустой **`cors_allowed_origins`**, шлюз для подходящего заголовка **`Origin`** добавляет заголовки CORS и отвечает на **`OPTIONS`** кодом **204** без тела (preflight). Совпадение: точное равенство строки origin или `"*"`. Whitelist IP по-прежнему применяется **до** обработки запроса.
|
||||||
|
|
||||||
## Ограничения
|
## Ограничения
|
||||||
|
|
||||||
- **Один и тот же `username` на разных серверах** может соответствовать разным учётным записям; агрегатор сопоставляет строки по имени — учитывайте при интерпретации сумм.
|
- **Один и тот же `username` на разных серверах** может соответствовать разным учётным записям; агрегатор сопоставляет строки по имени — учитывайте при интерпретации сумм.
|
||||||
- У Telemt в `UserInfo` **нет** поля «IP последний раз подключался к серверу X». В `unique-ips` поле `primary_server` заполняется **только** если ровно один сервер видит IP в `active_unique_ips_list` на момент запроса; иначе `primary_server` отсутствует или несколько серверов в списках — это снимок, не история.
|
- У 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
|
```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/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/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/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"
|
||||||
```
|
```
|
||||||
|
|||||||
@@ -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 }
|
||||||
+79
-95
@@ -15,105 +15,89 @@ import (
|
|||||||
|
|
||||||
const statsUsersPath = "stats/users"
|
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 {
|
func FetchStatsUsers(ctx context.Context, client *http.Client, parsed *config.Parsed, aliases []string) []ServerFetchResult {
|
||||||
out := make([]ServerFetchResult, 0, len(aliases))
|
out := make([]ServerFetchResult, 0, len(aliases))
|
||||||
for _, alias := range aliases {
|
for _, alias := range aliases {
|
||||||
srv := parsed.ByAlias[alias]
|
users, meta := FetchTelemtGET[[]UserInfo](ctx, client, parsed, alias, statsUsersPath)
|
||||||
if srv == nil {
|
fr := ServerFetchResult{
|
||||||
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{
|
|
||||||
Alias: alias,
|
Alias: alias,
|
||||||
OK: true,
|
OK: meta.OK,
|
||||||
HTTPStatus: resp.StatusCode,
|
HTTPStatus: meta.HTTPStatus,
|
||||||
LatencyMs: latency,
|
LatencyMs: meta.LatencyMs,
|
||||||
Revision: env.Revision,
|
Error: meta.Error,
|
||||||
Users: env.Data,
|
Revision: meta.Revision,
|
||||||
})
|
}
|
||||||
|
if meta.OK {
|
||||||
|
fr.Users = users
|
||||||
|
}
|
||||||
|
out = append(out, fr)
|
||||||
}
|
}
|
||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
+157
-20
@@ -4,8 +4,11 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"regexp"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/telemt/telemt-api/internal/config"
|
"github.com/telemt/telemt-api/internal/config"
|
||||||
@@ -14,17 +17,34 @@ import (
|
|||||||
|
|
||||||
const pathPrefix = "/api/agg"
|
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.
|
// Handler serves GET /api/agg/* aggregate endpoints.
|
||||||
type Handler struct {
|
type Handler struct {
|
||||||
Parsed *config.Parsed
|
Parsed *config.Parsed
|
||||||
Client *http.Client
|
Client *http.Client
|
||||||
Geo *geoip.Service
|
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).
|
// NewHandler builds an aggregate handler; client must use a non-nil Transport (e.g. gateway shared transport).
|
||||||
// Geo may be nil (no GeoLite2 lookups).
|
// Geo may be nil (no GeoLite2 lookups). cacheTTL 0 disables response caching.
|
||||||
func NewHandler(p *config.Parsed, client *http.Client, geo *geoip.Service) *Handler {
|
func NewHandler(p *config.Parsed, client *http.Client, geo *geoip.Service, cacheTTL time.Duration) *Handler {
|
||||||
return &Handler{Parsed: p, Client: client, Geo: geo}
|
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) {
|
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(r.URL.Path, pathPrefix)
|
||||||
sub = strings.TrimPrefix(sub, "/")
|
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)
|
h.handleSummary(w, r)
|
||||||
case "traffic":
|
case sub == "traffic":
|
||||||
h.handleTraffic(w, r)
|
h.handleTraffic(w, r)
|
||||||
case "unique-ips":
|
case sub == "unique-ips":
|
||||||
h.handleUniqueIPs(w, r)
|
h.handleUniqueIPs(w, r)
|
||||||
case "users":
|
case sub == "users":
|
||||||
h.handleUsers(w, r)
|
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:
|
default:
|
||||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
writeNotFound(w, "unknown aggregate path")
|
||||||
w.WriteHeader(http.StatusNotFound)
|
|
||||||
_ = json.NewEncoder(w).Encode(errEnvelope("not_found", "unknown aggregate path"))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -115,6 +183,15 @@ type resolveError struct {
|
|||||||
|
|
||||||
func (e *resolveError) Error() string { return e.msg }
|
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) {
|
func (h *Handler) handleSummary(w http.ResponseWriter, r *http.Request) {
|
||||||
aliases, err := h.resolveAliases(r)
|
aliases, err := h.resolveAliases(r)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -131,7 +208,8 @@ func (h *Handler) handleSummary(w http.ResponseWriter, r *http.Request) {
|
|||||||
defer cancel()
|
defer cancel()
|
||||||
results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
|
results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
|
||||||
data := BuildSummary(results, topN)
|
data := BuildSummary(results, topN)
|
||||||
writeOK(w, data)
|
partial := anyUpstreamFailed(results)
|
||||||
|
writeAggOK(w, partial, data)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) handleTraffic(w http.ResponseWriter, r *http.Request) {
|
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()
|
defer cancel()
|
||||||
results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
|
results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
|
||||||
data := BuildTraffic(results)
|
data := BuildTraffic(results)
|
||||||
writeOK(w, data)
|
partial := anyUpstreamFailed(results)
|
||||||
|
writeAggOK(w, partial, data)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) handleUniqueIPs(w http.ResponseWriter, r *http.Request) {
|
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") {
|
if h.Geo != nil && !strings.EqualFold(r.URL.Query().Get("geo"), "false") {
|
||||||
EnrichUniqueIPsGeo(data, h.Geo)
|
EnrichUniqueIPsGeo(data, h.Geo)
|
||||||
}
|
}
|
||||||
writeOK(w, data)
|
partial := anyUpstreamFailed(results)
|
||||||
|
writeAggOK(w, partial, data)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) handleUsers(w http.ResponseWriter, r *http.Request) {
|
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()
|
defer cancel()
|
||||||
results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
|
results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases)
|
||||||
data := BuildUsers(results, includeLinks, minOct)
|
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) {
|
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()))
|
_ = 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")
|
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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ func TestHandlerResolveAndFetch(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
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)
|
req := httptest.NewRequest(http.MethodGet, "/api/agg/summary?aliases=test", nil)
|
||||||
rec := httptest.NewRecorder()
|
rec := httptest.NewRecorder()
|
||||||
h.ServeHTTP(rec, req)
|
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) {
|
func TestHandlerMethodNotAllowed(t *testing.T) {
|
||||||
cfg := &config.Config{
|
cfg := &config.Config{
|
||||||
Servers: []config.Server{{Alias: "x", BaseURL: "http://127.0.0.1:1", PathPrefix: "/v1"}},
|
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 {
|
if err != nil {
|
||||||
t.Fatal(err)
|
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)
|
req := httptest.NewRequest(http.MethodPost, "/api/agg/summary", nil)
|
||||||
rec := httptest.NewRecorder()
|
rec := httptest.NewRecorder()
|
||||||
h.ServeHTTP(rec, req)
|
h.ServeHTTP(rec, req)
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package aggregate
|
|||||||
import (
|
import (
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
// BuildTraffic builds traffic matrix from successful fetches only.
|
// BuildTraffic builds traffic matrix from successful fetches only.
|
||||||
@@ -140,6 +141,11 @@ func sortedKeys(m map[string]struct{}) []string {
|
|||||||
return s
|
return s
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type userOnServer struct {
|
||||||
|
alias string
|
||||||
|
u UserInfo
|
||||||
|
}
|
||||||
|
|
||||||
// BuildUsers merged rows with totals and optional links from first successful server per user.
|
// BuildUsers merged rows with totals and optional links from first successful server per user.
|
||||||
func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets uint64) []UsersRow {
|
func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets uint64) []UsersRow {
|
||||||
type acc struct {
|
type acc struct {
|
||||||
@@ -150,6 +156,7 @@ func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets u
|
|||||||
rec uint64
|
rec uint64
|
||||||
}
|
}
|
||||||
m := map[string]*acc{}
|
m := map[string]*acc{}
|
||||||
|
perUserServers := map[string][]userOnServer{}
|
||||||
|
|
||||||
for _, fr := range results {
|
for _, fr := range results {
|
||||||
if !fr.OK {
|
if !fr.OK {
|
||||||
@@ -176,6 +183,7 @@ func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets u
|
|||||||
if includeLinks && u.Links != nil && a.links == nil {
|
if includeLinks && u.Links != nil && a.links == nil {
|
||||||
a.links = u.Links
|
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))
|
rows := make([]UsersRow, 0, len(names))
|
||||||
for _, name := range names {
|
for _, name := range names {
|
||||||
a := m[name]
|
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{
|
rows = append(rows, UsersRow{
|
||||||
Username: name,
|
Username: name,
|
||||||
TotalMegabytes: octetsToMegabytes(a.total),
|
TotalMegabytes: octetsToMegabytes(a.total),
|
||||||
ByServer: a.byServer,
|
ByServer: a.byServer,
|
||||||
Links: a.links,
|
Links: a.links,
|
||||||
ActiveUniqueIPs: a.act,
|
ActiveUniqueIPs: a.act,
|
||||||
RecentUniqueIPs: a.rec,
|
RecentUniqueIPs: a.rec,
|
||||||
|
UserAdTag: ad,
|
||||||
|
MaxTCPConns: mt,
|
||||||
|
ExpirationRFC3339: ex,
|
||||||
|
DataQuotaBytes: dq,
|
||||||
|
MaxUniqueIPs: mu,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
return rows
|
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.
|
// BuildSummary computes fleet totals and top users.
|
||||||
func BuildSummary(results []ServerFetchResult, topN int) SummaryData {
|
func BuildSummary(results []ServerFetchResult, topN int) SummaryData {
|
||||||
if topN <= 0 {
|
if topN <= 0 {
|
||||||
|
|||||||
@@ -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) {
|
func TestBuildUsersMinOctets(t *testing.T) {
|
||||||
results := []ServerFetchResult{
|
results := []ServerFetchResult{
|
||||||
{Alias: "a", OK: true, Users: []UserInfo{{Username: "low", TotalOctets: 5}}},
|
{Alias: "a", OK: true, Users: []UserInfo{{Username: "low", TotalOctets: 5}}},
|
||||||
|
|||||||
@@ -24,10 +24,55 @@ type UserLinks struct {
|
|||||||
TLS []string `json:"tls,omitempty"`
|
TLS []string `json:"tls,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type statsUsersEnvelope struct {
|
// HealthData mirrors Telemt GET /v1/health data block.
|
||||||
OK bool `json:"ok"`
|
type HealthData struct {
|
||||||
Data []UserInfo `json:"data"`
|
Status string `json:"status"`
|
||||||
Revision string `json:"revision"`
|
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.
|
// 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.
|
// UsersRow merged user view with optional per-server detail and links.
|
||||||
type UsersRow struct {
|
type UsersRow struct {
|
||||||
Username string `json:"username"`
|
Username string `json:"username"`
|
||||||
TotalMegabytes float64 `json:"total_megabytes"`
|
TotalMegabytes float64 `json:"total_megabytes"`
|
||||||
ByServer map[string]TrafficServerStats `json:"by_server"`
|
ByServer map[string]TrafficServerStats `json:"by_server"`
|
||||||
Links *UserLinks `json:"links,omitempty"`
|
Links *UserLinks `json:"links,omitempty"`
|
||||||
ActiveUniqueIPs uint64 `json:"active_unique_ips"`
|
ActiveUniqueIPs uint64 `json:"active_unique_ips"`
|
||||||
RecentUniqueIPs uint64 `json:"recent_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.
|
// SummaryData fleet snapshot.
|
||||||
|
|||||||
@@ -15,13 +15,14 @@ var aliasRe = regexp.MustCompile(`^[a-z0-9][a-z0-9_-]*$`)
|
|||||||
|
|
||||||
// Config is the gateway YAML configuration.
|
// Config is the gateway YAML configuration.
|
||||||
type Config struct {
|
type Config struct {
|
||||||
Listen string `yaml:"listen"`
|
Listen string `yaml:"listen"`
|
||||||
AllowAll bool `yaml:"allow_all"`
|
AllowAll bool `yaml:"allow_all"`
|
||||||
WhitelistCIDRs []string `yaml:"whitelist_cidrs"`
|
WhitelistCIDRs []string `yaml:"whitelist_cidrs"`
|
||||||
TrustedProxies []string `yaml:"trusted_proxies"`
|
TrustedProxies []string `yaml:"trusted_proxies"`
|
||||||
Servers []Server `yaml:"servers"`
|
CorsAllowedOrigins []string `yaml:"cors_allowed_origins"`
|
||||||
Aggregate *AggregateConfig `yaml:"aggregate"`
|
Servers []Server `yaml:"servers"`
|
||||||
GeoIP *GeoIPConfig `yaml:"geoip"`
|
Aggregate *AggregateConfig `yaml:"aggregate"`
|
||||||
|
GeoIP *GeoIPConfig `yaml:"geoip"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// GeoIPConfig enables GeoLite2 lookups for /api/agg/unique-ips (optional).
|
// GeoIPConfig enables GeoLite2 lookups for /api/agg/unique-ips (optional).
|
||||||
@@ -38,6 +39,8 @@ type GeoIPConfig struct {
|
|||||||
type AggregateConfig struct {
|
type AggregateConfig struct {
|
||||||
// IncludeAliases limits aggregation to these server aliases; empty means all servers.
|
// IncludeAliases limits aggregation to these server aliases; empty means all servers.
|
||||||
IncludeAliases []string `yaml:"include_aliases"`
|
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.
|
// 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)
|
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
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -98,3 +98,19 @@ func TestValidateAggregateIncludeAliases(t *testing.T) {
|
|||||||
t.Fatal("expected error for unknown include alias")
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ type Gateway struct {
|
|||||||
log *slog.Logger
|
log *slog.Logger
|
||||||
transport *http.Transport
|
transport *http.Transport
|
||||||
promHandler http.Handler
|
promHandler http.Handler
|
||||||
|
corsAllowed []string
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewGateway builds handlers and reverse proxies from parsed config.
|
// 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.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
|
return g, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Handler returns the root HTTP handler with middleware.
|
// Handler returns the root HTTP handler with middleware.
|
||||||
func (g *Gateway) Handler() http.Handler {
|
func (g *Gateway) Handler() http.Handler {
|
||||||
var h http.Handler = http.HandlerFunc(g.serve)
|
var h http.Handler = http.HandlerFunc(g.serve)
|
||||||
|
h = g.withCORS(h)
|
||||||
h = g.withWhitelist(h)
|
h = g.withWhitelist(h)
|
||||||
h = g.withAccessLog(h)
|
h = g.withAccessLog(h)
|
||||||
h = g.withMetrics(h)
|
h = g.withMetrics(h)
|
||||||
return 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 {
|
func (g *Gateway) withWhitelist(next http.Handler) http.Handler {
|
||||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
if r.URL.Path == "/health" {
|
if r.URL.Path == "/health" {
|
||||||
|
|||||||
Reference in New Issue
Block a user