diff --git a/README.md b/README.md index f9cfef7..518bda8 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # telemt-api -HTTP‑шлюз на Go для [Telemt Control API](docs/API.md): один порт, **белый список IP (CIDR)**, маршруты вида `/api/{alias}/…` → `{base_url}/v1/…`, метрики Prometheus на `/metrics`. +HTTP‑шлюз на Go для [Telemt Control API](docs/API.md): один порт, **белый список IP (CIDR)**, маршруты вида `/api/{alias}/…` → `{base_url}/v1/…`, агрегация нескольких инстансов — [`/api/agg/…`](docs/AGGREGATE.md), метрики Prometheus на `/metrics`. ## Быстрый старт (Linux) @@ -55,6 +55,7 @@ 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/…`) | ## Сборка и тесты без Docker diff --git a/config.example.yaml b/config.example.yaml index c021f29..1ea7bd1 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -23,6 +23,12 @@ whitelist_cidrs: # - "10.0.0.0/8" trusted_proxies: [] +# Опционально: по умолчанию /api/agg/* опрашивает все servers; можно ограничить список: +# aggregate: +# include_aliases: +# - gt1 +# - gt2 + servers: - alias: main_srv base_url: http://127.0.0.1:9091 diff --git a/docs/AGGREGATE.md b/docs/AGGREGATE.md new file mode 100644 index 0000000..68f4cee --- /dev/null +++ b/docs/AGGREGATE.md @@ -0,0 +1,52 @@ +# Агрегирующие эндпоинты шлюза (`/api/agg/`) + +Шлюз **telemt-api** опрашивает несколько upstream [Telemt Control API](API.md) (`GET /v1/stats/users` на каждом сервере из конфигурации) и отдаёт сводные JSON-ответы в формате `{"ok": true, "data": ...}`. + +Доступ к **одному** инстансу по-прежнему через прокси: `GET /api/{alias}/…` (например `/api/gt1/v1/stats/users`). + +## Маршруты + +| Метод | Путь | Описание | +| --- | --- | --- | +| GET | `/api/agg/summary` | Сводка по флоту, список опросов upstream, топ пользователей по суммарному `total_octets`. | +| GET | `/api/agg/traffic` | Трафик по каждому пользователю в разрезе серверов (алиасов). | +| GET | `/api/agg/unique-ips` | Уникальные IP по пользователю: на каких серверах IP есть в active/recent списках снимка. | +| GET | `/api/agg/users` | Объединённый список пользователей с `by_server` и суммарным `total_octets`. | + +Все методы — **GET**; действует тот же whitelist, что и для остального API шлюза. + +## Query-параметры + +| Параметр | Где | Значение | +| --- | --- | --- | +| `aliases` | все | Список алиасов через запятую (например `gt1,gt2`). Если не задан — см. `aggregate.include_aliases` в YAML или все серверы из `servers`. | +| `top_n` | `summary` | Размер топа пользователей (по умолчанию `10`, максимум `1000`). | +| `include_links` | `users` | `true` — добавить сгенерированные `tg://proxy` ссылки (берётся первая успешная запись по пользователю). | +| `min_total_octets` | `users` | Отфильтровать пользователей с суммарным трафиком ниже порога. | + +## Конфигурация (опционально) + +```yaml +aggregate: + include_aliases: + - gt1 + - gt2 +``` + +Если блок отсутствует или `include_aliases` пуст, по умолчанию участвуют **все** записи `servers`. + +Имя алиаса **`agg`** в `servers` запрещено (зарезервировано под префикс `/api/agg/`). + +## Ограничения + +- **Один и тот же `username` на разных серверах** может соответствовать разным учётным записям; агрегатор сопоставляет строки по имени — учитывайте при интерпретации сумм. +- У Telemt в `UserInfo` **нет** поля «IP последний раз подключался к серверу X». В `unique-ips` поле `primary_server` заполняется **только** если ровно один сервер видит IP в `active_unique_ips_list` на момент запроса; иначе `primary_server` отсутствует или несколько серверов в списках — это снимок, не история. + +## Примеры + +```bash +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_octets=1000000" +``` diff --git a/internal/aggregate/fetch.go b/internal/aggregate/fetch.go new file mode 100644 index 0000000..4a53185 --- /dev/null +++ b/internal/aggregate/fetch.go @@ -0,0 +1,148 @@ +package aggregate + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/telemt/telemt-api/internal/config" +) + +const statsUsersPath = "stats/users" + +// FetchStatsUsers calls GET {base}{path_prefix}/stats/users for each alias. +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{ + Alias: alias, + OK: true, + HTTPStatus: resp.StatusCode, + LatencyMs: latency, + Revision: env.Revision, + Users: env.Data, + }) + } + return out +} + +func joinPathPrefix(base *url.URL, pathPrefix, rest string) *url.URL { + rel := strings.Trim(pathPrefix, "/") + if rest != "" { + if rel != "" { + rel = rel + "/" + rest + } else { + rel = rest + } + } + var parts []string + for _, seg := range strings.Split(rel, "/") { + if seg != "" { + parts = append(parts, seg) + } + } + if len(parts) == 0 { + out := *base + return &out + } + return base.JoinPath(parts...) +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/internal/aggregate/handlers.go b/internal/aggregate/handlers.go new file mode 100644 index 0000000..71c6b4f --- /dev/null +++ b/internal/aggregate/handlers.go @@ -0,0 +1,189 @@ +package aggregate + +import ( + "context" + "encoding/json" + "net/http" + "strconv" + "strings" + "time" + + "github.com/telemt/telemt-api/internal/config" +) + +const pathPrefix = "/api/agg" + +// Handler serves GET /api/agg/* aggregate endpoints. +type Handler struct { + Parsed *config.Parsed + Client *http.Client +} + +// NewHandler builds an aggregate handler; client must use a non-nil Transport (e.g. gateway shared transport). +func NewHandler(p *config.Parsed, client *http.Client) *Handler { + return &Handler{Parsed: p, Client: client} +} + +func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(http.StatusMethodNotAllowed) + _ = json.NewEncoder(w).Encode(errEnvelope("method_not_allowed", "only GET is allowed")) + return + } + sub := strings.TrimPrefix(r.URL.Path, pathPrefix) + sub = strings.TrimPrefix(sub, "/") + switch sub { + case "summary": + h.handleSummary(w, r) + case "traffic": + h.handleTraffic(w, r) + case "unique-ips": + h.handleUniqueIPs(w, r) + case "users": + h.handleUsers(w, r) + 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")) + } +} + +func errEnvelope(code, msg string) map[string]any { + return map[string]any{ + "ok": false, + "error": map[string]string{ + "code": code, + "message": msg, + }, + } +} + +func (h *Handler) resolveAliases(r *http.Request) ([]string, error) { + q := r.URL.Query().Get("aliases") + if strings.TrimSpace(q) != "" { + var out []string + for _, p := range strings.Split(q, ",") { + a := strings.TrimSpace(p) + if a == "" { + continue + } + if h.Parsed.ByAlias[a] == nil { + return nil, &resolveError{msg: "unknown alias: " + a} + } + out = append(out, a) + } + if len(out) == 0 { + return nil, &resolveError{msg: "aliases query produced empty list"} + } + return out, nil + } + cfg := h.Parsed.Config.Aggregate + if cfg != nil && len(cfg.IncludeAliases) > 0 { + for _, a := range cfg.IncludeAliases { + a = strings.TrimSpace(a) + if a == "" { + continue + } + if h.Parsed.ByAlias[a] == nil { + return nil, &resolveError{msg: "aggregate.include_aliases: unknown alias: " + a} + } + } + out := make([]string, 0, len(cfg.IncludeAliases)) + for _, a := range cfg.IncludeAliases { + a = strings.TrimSpace(a) + if a == "" { + continue + } + out = append(out, a) + } + return out, nil + } + out := make([]string, 0, len(h.Parsed.Config.Servers)) + for i := range h.Parsed.Config.Servers { + out = append(out, h.Parsed.Config.Servers[i].Alias) + } + return out, nil +} + +type resolveError struct { + msg string +} + +func (e *resolveError) Error() string { return e.msg } + +func (h *Handler) handleSummary(w http.ResponseWriter, r *http.Request) { + aliases, err := h.resolveAliases(r) + if err != nil { + writeBadRequest(w, err) + return + } + topN := 10 + if v := r.URL.Query().Get("top_n"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + topN = n + } + } + ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second) + defer cancel() + results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases) + data := BuildSummary(results, topN) + writeOK(w, data) +} + +func (h *Handler) handleTraffic(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() + results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases) + data := BuildTraffic(results) + writeOK(w, data) +} + +func (h *Handler) handleUniqueIPs(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() + results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases) + data := BuildUniqueIPs(results) + writeOK(w, data) +} + +func (h *Handler) handleUsers(w http.ResponseWriter, r *http.Request) { + aliases, err := h.resolveAliases(r) + if err != nil { + writeBadRequest(w, err) + return + } + includeLinks := strings.EqualFold(r.URL.Query().Get("include_links"), "true") + minOct := uint64(0) + if v := r.URL.Query().Get("min_total_octets"); v != "" { + if n, err := strconv.ParseUint(v, 10, 64); err == nil { + minOct = n + } + } + ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second) + defer cancel() + results := FetchStatsUsers(ctx, h.Client, h.Parsed, aliases) + data := BuildUsers(results, includeLinks, minOct) + writeOK(w, data) +} + +func writeBadRequest(w http.ResponseWriter, err error) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(errEnvelope("bad_request", err.Error())) +} + +func writeOK(w http.ResponseWriter, data any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "data": data}) +} diff --git a/internal/aggregate/handlers_test.go b/internal/aggregate/handlers_test.go new file mode 100644 index 0000000..0b322d5 --- /dev/null +++ b/internal/aggregate/handlers_test.go @@ -0,0 +1,75 @@ +package aggregate + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/telemt/telemt-api/internal/config" +) + +func TestHandlerResolveAndFetch(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": 42, "current_connections": 0}}, + "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()) + req := httptest.NewRequest(http.MethodGet, "/api/agg/summary?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 SummaryData `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil { + t.Fatal(err) + } + if !env.OK || env.Data.FleetTotalOctets != 42 { + t.Fatalf("data: %+v", env.Data) + } +} + +func TestHandlerMethodNotAllowed(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) + req := httptest.NewRequest(http.MethodPost, "/api/agg/summary", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("got %d", rec.Code) + } +} diff --git a/internal/aggregate/merge.go b/internal/aggregate/merge.go new file mode 100644 index 0000000..e9ae00a --- /dev/null +++ b/internal/aggregate/merge.go @@ -0,0 +1,272 @@ +package aggregate + +import ( + "sort" + "strings" +) + +// BuildTraffic builds traffic matrix from successful fetches only. +func BuildTraffic(results []ServerFetchResult) []TrafficRow { + perUser := map[string]map[string]TrafficServerStats{} + + for _, fr := range results { + if !fr.OK { + continue + } + for _, u := range fr.Users { + if perUser[u.Username] == nil { + perUser[u.Username] = map[string]TrafficServerStats{} + } + perUser[u.Username][fr.Alias] = TrafficServerStats{ + TotalOctets: u.TotalOctets, + CurrentConnections: u.CurrentConnections, + Revision: fr.Revision, + } + } + } + + names := make([]string, 0, len(perUser)) + for n := range perUser { + names = append(names, n) + } + sort.Strings(names) + + rows := make([]TrafficRow, 0, len(names)) + for _, name := range names { + rows = append(rows, TrafficRow{ + Username: name, + Servers: perUser[name], + }) + } + return rows +} + +// BuildUniqueIPs builds IP × server visibility per user. +func BuildUniqueIPs(results []ServerFetchResult) []UniqueIPsRow { + type ipKey struct { + user string + ip string + } + active := map[ipKey]map[string]struct{}{} + recent := map[ipKey]map[string]struct{}{} + + for _, fr := range results { + if !fr.OK { + continue + } + for _, u := range fr.Users { + for _, ip := range u.ActiveUniqueIPsList { + ip = strings.TrimSpace(ip) + if ip == "" { + continue + } + k := ipKey{user: u.Username, ip: ip} + if active[k] == nil { + active[k] = map[string]struct{}{} + } + active[k][fr.Alias] = struct{}{} + } + for _, ip := range u.RecentUniqueIPsList { + ip = strings.TrimSpace(ip) + if ip == "" { + continue + } + k := ipKey{user: u.Username, ip: ip} + if recent[k] == nil { + recent[k] = map[string]struct{}{} + } + recent[k][fr.Alias] = struct{}{} + } + } + } + + users := map[string][]ipKey{} + for k := range active { + users[k.user] = append(users[k.user], k) + } + for k := range recent { + found := false + for _, x := range users[k.user] { + if x == k { + found = true + break + } + } + if !found { + users[k.user] = append(users[k.user], k) + } + } + + userNames := make([]string, 0, len(users)) + for u := range users { + userNames = append(userNames, u) + } + sort.Strings(userNames) + + out := make([]UniqueIPsRow, 0, len(userNames)) + for _, uname := range userNames { + keys := users[uname] + sort.Slice(keys, func(i, j int) bool { return keys[i].ip < keys[j].ip }) + ips := make([]IPAssignments, 0, len(keys)) + for _, k := range keys { + a := sortedKeys(active[k]) + r := sortedKeys(recent[k]) + var primary *string + if len(a) == 1 { + s := a[0] + primary = &s + } + ips = append(ips, IPAssignments{ + IP: k.ip, + ActiveOnServers: a, + RecentOnServers: r, + PrimaryServer: primary, + }) + } + out = append(out, UniqueIPsRow{Username: uname, IPs: ips}) + } + return out +} + +func sortedKeys(m map[string]struct{}) []string { + if len(m) == 0 { + return nil + } + s := make([]string, 0, len(m)) + for k := range m { + s = append(s, k) + } + sort.Strings(s) + return s +} + +// 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 { + byServer map[string]TrafficServerStats + links *UserLinks + total uint64 + act uint64 + rec uint64 + } + m := map[string]*acc{} + + for _, fr := range results { + if !fr.OK { + continue + } + for _, u := range fr.Users { + a := m[u.Username] + if a == nil { + a = &acc{byServer: map[string]TrafficServerStats{}} + m[u.Username] = a + } + a.byServer[fr.Alias] = TrafficServerStats{ + TotalOctets: u.TotalOctets, + CurrentConnections: u.CurrentConnections, + Revision: fr.Revision, + } + a.total += u.TotalOctets + if u.ActiveUniqueIPs > a.act { + a.act = u.ActiveUniqueIPs + } + if u.RecentUniqueIPs > a.rec { + a.rec = u.RecentUniqueIPs + } + if includeLinks && u.Links != nil && a.links == nil { + a.links = u.Links + } + } + } + + names := make([]string, 0, len(m)) + for n := range m { + if m[n].total >= minTotalOctets { + names = append(names, n) + } + } + sort.Strings(names) + + rows := make([]UsersRow, 0, len(names)) + for _, name := range names { + a := m[name] + rows = append(rows, UsersRow{ + Username: name, + TotalOctets: a.total, + ByServer: a.byServer, + Links: a.links, + ActiveUniqueIPs: a.act, + RecentUniqueIPs: a.rec, + }) + } + return rows +} + +// BuildSummary computes fleet totals and top users. +func BuildSummary(results []ServerFetchResult, topN int) SummaryData { + if topN <= 0 { + topN = 10 + } + if topN > 1000 { + topN = 1000 + } + + sumByUser := map[string]uint64{} + var fleetOctets, fleetConn uint64 + ok, fail := 0, 0 + serverRows := make([]ServerFetchResult, 0, len(results)) + + for _, fr := range results { + sr := ServerFetchResult{ + Alias: fr.Alias, + OK: fr.OK, + HTTPStatus: fr.HTTPStatus, + LatencyMs: fr.LatencyMs, + Error: fr.Error, + Revision: fr.Revision, + } + serverRows = append(serverRows, sr) + if !fr.OK { + fail++ + continue + } + ok++ + for _, u := range fr.Users { + fleetOctets += u.TotalOctets + fleetConn += u.CurrentConnections + sumByUser[u.Username] += u.TotalOctets + } + } + + type pair struct { + name string + n uint64 + } + pairs := make([]pair, 0, len(sumByUser)) + for n, v := range sumByUser { + pairs = append(pairs, pair{name: n, n: v}) + } + sort.Slice(pairs, func(i, j int) bool { + if pairs[i].n != pairs[j].n { + return pairs[i].n > pairs[j].n + } + return pairs[i].name < pairs[j].name + }) + if len(pairs) > topN { + pairs = pairs[:topN] + } + top := make([]TopUser, 0, len(pairs)) + for _, p := range pairs { + top = append(top, TopUser{Username: p.name, TotalOctets: p.n}) + } + + return SummaryData{ + Servers: serverRows, + ServersTotal: len(results), + ServersOK: ok, + ServersFailed: fail, + FleetTotalOctets: fleetOctets, + FleetTotalConnections: fleetConn, + TopUsers: top, + } +} diff --git a/internal/aggregate/merge_test.go b/internal/aggregate/merge_test.go new file mode 100644 index 0000000..431da18 --- /dev/null +++ b/internal/aggregate/merge_test.go @@ -0,0 +1,122 @@ +package aggregate + +import ( + "reflect" + "testing" +) + +func TestBuildTraffic(t *testing.T) { + results := []ServerFetchResult{ + { + Alias: "a", OK: true, Revision: "r1", + Users: []UserInfo{ + {Username: "u1", TotalOctets: 100, CurrentConnections: 2}, + {Username: "u2", TotalOctets: 50, CurrentConnections: 0}, + }, + }, + { + Alias: "b", OK: true, Revision: "r2", + Users: []UserInfo{ + {Username: "u1", TotalOctets: 30, CurrentConnections: 1}, + }, + }, + } + rows := BuildTraffic(results) + if len(rows) != 2 { + t.Fatalf("len rows: %d", len(rows)) + } + var u1 *TrafficRow + for i := range rows { + if rows[i].Username == "u1" { + u1 = &rows[i] + break + } + } + if u1 == nil { + t.Fatal("missing u1") + } + if u1.Servers["a"].TotalOctets != 100 || u1.Servers["b"].TotalOctets != 30 { + t.Fatalf("u1 servers: %+v", u1.Servers) + } +} + +func TestBuildUniqueIPs(t *testing.T) { + results := []ServerFetchResult{ + { + Alias: "s1", OK: true, + Users: []UserInfo{ + { + Username: "alice", + ActiveUniqueIPsList: []string{"10.0.0.1"}, + RecentUniqueIPsList: []string{"10.0.0.2"}, + }, + }, + }, + { + Alias: "s2", OK: true, + Users: []UserInfo{ + { + Username: "alice", + ActiveUniqueIPsList: []string{"10.0.0.1", "10.0.0.3"}, + }, + }, + }, + } + rows := BuildUniqueIPs(results) + if len(rows) != 1 || rows[0].Username != "alice" { + t.Fatalf("rows: %+v", rows) + } + var ip1 *IPAssignments + for i := range rows[0].IPs { + if rows[0].IPs[i].IP == "10.0.0.1" { + ip1 = &rows[0].IPs[i] + break + } + } + if ip1 == nil { + t.Fatal("missing 10.0.0.1") + } + if !reflect.DeepEqual(ip1.ActiveOnServers, []string{"s1", "s2"}) { + t.Fatalf("active: %v", ip1.ActiveOnServers) + } + if ip1.PrimaryServer != nil { + t.Fatalf("expected no primary, got %v", *ip1.PrimaryServer) + } +} + +func TestBuildUniqueIPsPrimary(t *testing.T) { + results := []ServerFetchResult{ + {Alias: "only", OK: true, Users: []UserInfo{ + {Username: "bob", ActiveUniqueIPsList: []string{"192.168.1.1"}}, + }}, + } + rows := BuildUniqueIPs(results) + if len(rows[0].IPs) != 1 || rows[0].IPs[0].PrimaryServer == nil || *rows[0].IPs[0].PrimaryServer != "only" { + t.Fatalf("primary: %+v", rows[0].IPs) + } +} + +func TestBuildSummary(t *testing.T) { + results := []ServerFetchResult{ + {Alias: "a", OK: true, Users: []UserInfo{{Username: "x", TotalOctets: 100, CurrentConnections: 2}}}, + {Alias: "b", OK: false, Error: "down"}, + } + s := BuildSummary(results, 5) + if s.ServersOK != 1 || s.ServersFailed != 1 || s.FleetTotalOctets != 100 || s.FleetTotalConnections != 2 { + t.Fatalf("summary: %+v", s) + } + if len(s.TopUsers) != 1 || s.TopUsers[0].Username != "x" { + t.Fatalf("top: %+v", s.TopUsers) + } +} + +func TestBuildUsersMinOctets(t *testing.T) { + results := []ServerFetchResult{ + {Alias: "a", OK: true, Users: []UserInfo{{Username: "low", TotalOctets: 5}}}, + {Alias: "b", OK: true, Users: []UserInfo{{Username: "high", TotalOctets: 100}}}, + } + rows := BuildUsers(results, false, 50) + if len(rows) != 1 || rows[0].Username != "high" { + t.Fatalf("rows: %+v", rows) + } +} diff --git a/internal/aggregate/types.go b/internal/aggregate/types.go new file mode 100644 index 0000000..96173d0 --- /dev/null +++ b/internal/aggregate/types.go @@ -0,0 +1,96 @@ +package aggregate + +// UserInfo mirrors Telemt Control API /v1/stats/users items (subset used by aggregator). +type UserInfo struct { + Username string `json:"username"` + UserAdTag *string `json:"user_ad_tag"` + MaxTCPConns *uint64 `json:"max_tcp_conns"` + ExpirationRFC3339 *string `json:"expiration_rfc3339"` + DataQuotaBytes *uint64 `json:"data_quota_bytes"` + MaxUniqueIPs *uint64 `json:"max_unique_ips"` + CurrentConnections uint64 `json:"current_connections"` + ActiveUniqueIPs uint64 `json:"active_unique_ips"` + ActiveUniqueIPsList []string `json:"active_unique_ips_list"` + RecentUniqueIPs uint64 `json:"recent_unique_ips"` + RecentUniqueIPsList []string `json:"recent_unique_ips_list"` + TotalOctets uint64 `json:"total_octets"` + Links *UserLinks `json:"links,omitempty"` +} + +// UserLinks optional tg://proxy link groups. +type UserLinks struct { + Classic []string `json:"classic,omitempty"` + Secure []string `json:"secure,omitempty"` + TLS []string `json:"tls,omitempty"` +} + +type statsUsersEnvelope struct { + OK bool `json:"ok"` + Data []UserInfo `json:"data"` + Revision string `json:"revision"` +} + +// ServerFetchResult is one upstream GET /v1/stats/users outcome. +type ServerFetchResult struct { + Alias string `json:"alias"` + 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"` + Users []UserInfo `json:"-"` +} + +// TrafficRow is per-user traffic broken down by server alias. +type TrafficRow struct { + Username string `json:"username"` + Servers map[string]TrafficServerStats `json:"servers"` +} + +// TrafficServerStats per-server counters for one user. +type TrafficServerStats struct { + TotalOctets uint64 `json:"total_octets"` + CurrentConnections uint64 `json:"current_connections"` + Revision string `json:"revision,omitempty"` +} + +// UniqueIPsRow describes IP presence across servers for one user. +type UniqueIPsRow struct { + Username string `json:"username"` + IPs []IPAssignments `json:"ips"` +} + +// IPAssignments per-IP server visibility (snapshot-based). +type IPAssignments struct { + IP string `json:"ip"` + ActiveOnServers []string `json:"active_on_servers"` + RecentOnServers []string `json:"recent_on_servers"` + PrimaryServer *string `json:"primary_server,omitempty"` +} + +// UsersRow merged user view with optional per-server detail and links. +type UsersRow struct { + Username string `json:"username"` + TotalOctets uint64 `json:"total_octets"` + ByServer map[string]TrafficServerStats `json:"by_server"` + Links *UserLinks `json:"links,omitempty"` + ActiveUniqueIPs uint64 `json:"active_unique_ips"` + RecentUniqueIPs uint64 `json:"recent_unique_ips"` +} + +// SummaryData fleet snapshot. +type SummaryData struct { + Servers []ServerFetchResult `json:"servers"` + ServersTotal int `json:"servers_total"` + ServersOK int `json:"servers_ok"` + ServersFailed int `json:"servers_failed"` + FleetTotalOctets uint64 `json:"fleet_total_octets"` + FleetTotalConnections uint64 `json:"fleet_total_connections"` + TopUsers []TopUser `json:"top_users"` +} + +// TopUser by summed total_octets across servers for one username. +type TopUser struct { + Username string `json:"username"` + TotalOctets uint64 `json:"total_octets"` +} diff --git a/internal/config/config.go b/internal/config/config.go index ed79a3b..850e17d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -15,11 +15,18 @@ 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"` + 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"` +} + +// AggregateConfig controls default scope of /api/agg/* (optional). +type AggregateConfig struct { + // IncludeAliases limits aggregation to these server aliases; empty means all servers. + IncludeAliases []string `yaml:"include_aliases"` } // Server maps a URL alias to an upstream base URL. @@ -63,6 +70,9 @@ func (c *Config) Validate() error { if _, ok := seen[s.Alias]; ok { return fmt.Errorf("duplicate alias %q", s.Alias) } + if s.Alias == "agg" { + return fmt.Errorf("servers[%d]: alias %q is reserved for /api/agg/", i, s.Alias) + } seen[s.Alias] = struct{}{} if s.BaseURL == "" { return fmt.Errorf("servers[%d]: base_url is required", i) @@ -92,6 +102,17 @@ func (c *Config) Validate() error { return fmt.Errorf("trusted_proxies[%d]: %w", i, err) } } + if c.Aggregate != nil { + for i, a := range c.Aggregate.IncludeAliases { + a = strings.TrimSpace(a) + if a == "" { + return fmt.Errorf("aggregate.include_aliases[%d]: empty entry", i) + } + if !seen[a] { + return fmt.Errorf("aggregate.include_aliases[%d]: unknown server alias %q", i, a) + } + } + } return nil } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 617d755..5c73f23 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -62,3 +62,39 @@ func TestValidateDuplicateAlias(t *testing.T) { t.Fatal("expected error") } } + +func TestValidateReservedAggAlias(t *testing.T) { + c := &Config{ + Servers: []Server{ + {Alias: "agg", BaseURL: "http://x:1"}, + }, + } + if err := c.Validate(); err == nil { + t.Fatal("expected error for reserved alias agg") + } +} + +func TestValidateAggregateIncludeAliases(t *testing.T) { + c := &Config{ + Servers: []Server{ + {Alias: "a", BaseURL: "http://x:1"}, + }, + Aggregate: &AggregateConfig{ + IncludeAliases: []string{"a"}, + }, + } + if err := c.Validate(); err != nil { + t.Fatal(err) + } + c2 := &Config{ + Servers: []Server{ + {Alias: "a", BaseURL: "http://x:1"}, + }, + Aggregate: &AggregateConfig{ + IncludeAliases: []string{"nope"}, + }, + } + if err := c2.Validate(); err == nil { + t.Fatal("expected error for unknown include alias") + } +} diff --git a/internal/server/gateway.go b/internal/server/gateway.go index dd46191..2e8d8b2 100644 --- a/internal/server/gateway.go +++ b/internal/server/gateway.go @@ -13,6 +13,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/telemt/telemt-api/internal/aggregate" "github.com/telemt/telemt-api/internal/config" "github.com/telemt/telemt-api/internal/proxy" ) @@ -21,6 +22,7 @@ import ( type Gateway struct { parsed *config.Parsed proxies map[string]*httputil.ReverseProxy + agg *aggregate.Handler log *slog.Logger transport *http.Transport promHandler http.Handler @@ -63,6 +65,7 @@ func NewGateway(p *config.Parsed, log *slog.Logger) (*Gateway, error) { } g.proxies[s.Alias] = rp } + g.agg = aggregate.NewHandler(p, &http.Client{Transport: t}) return g, nil } @@ -170,6 +173,10 @@ func (g *Gateway) serve(w http.ResponseWriter, r *http.Request) { return } const prefix = "/api/" + if r.URL.Path == "/api/agg" || strings.HasPrefix(r.URL.Path, "/api/agg/") { + g.agg.ServeHTTP(w, r) + return + } if !strings.HasPrefix(r.URL.Path, prefix) { http.NotFound(w, r) return