diff --git a/docs/AGGREGATE.md b/docs/AGGREGATE.md index 68f4cee..64be5bd 100644 --- a/docs/AGGREGATE.md +++ b/docs/AGGREGATE.md @@ -2,16 +2,18 @@ Шлюз **telemt-api** опрашивает несколько upstream [Telemt Control API](API.md) (`GET /v1/stats/users` на каждом сервере из конфигурации) и отдаёт сводные JSON-ответы в формате `{"ok": true, "data": ...}`. -Доступ к **одному** инстансу по-прежнему через прокси: `GET /api/{alias}/…` (например `/api/gt1/v1/stats/users`). +**Единицы трафика в агрегатах:** поля `*_megabytes` — это **двоичные мегабайты (MiB)**, 1 MiB = 1024² октетов (как у Telemt в ответе считаются октеты, шлюз делит на MiB для удобства). + +Доступ к **одному** инстансу по-прежнему через прокси: `GET /api/{alias}/…` (например `/api/gt1/v1/stats/users`) — там по-прежнему `total_octets` как в [API.md](API.md). ## Маршруты | Метод | Путь | Описание | | --- | --- | --- | -| GET | `/api/agg/summary` | Сводка по флоту, список опросов upstream, топ пользователей по суммарному `total_octets`. | -| GET | `/api/agg/traffic` | Трафик по каждому пользователю в разрезе серверов (алиасов). | +| GET | `/api/agg/summary` | Сводка по флоту, список опросов upstream, `fleet_total_megabytes` / `fleet_total_connections`. Два топа (размер задаётся `top_n`): **`top_users`** — самые «прожорливые» по суммарному трафику (MiB) по всем серверам; **`top_users_by_unique_ips`** — по максимальному `active_unique_ips` среди серверов для пользователя (как в Telemt, снимок). | +| GET | `/api/agg/traffic` | Трафик по каждому пользователю в разрезе серверов: `servers..total_megabytes`. | | GET | `/api/agg/unique-ips` | Уникальные IP по пользователю: на каких серверах IP есть в active/recent списках снимка. | -| GET | `/api/agg/users` | Объединённый список пользователей с `by_server` и суммарным `total_octets`. | +| GET | `/api/agg/users` | Объединённый список пользователей с `by_server` и суммарным `total_megabytes`. | Все методы — **GET**; действует тот же whitelist, что и для остального API шлюза. @@ -22,7 +24,8 @@ | `aliases` | все | Список алиасов через запятую (например `gt1,gt2`). Если не задан — см. `aggregate.include_aliases` в YAML или все серверы из `servers`. | | `top_n` | `summary` | Размер топа пользователей (по умолчанию `10`, максимум `1000`). | | `include_links` | `users` | `true` — добавить сгенерированные `tg://proxy` ссылки (берётся первая успешная запись по пользователю). | -| `min_total_octets` | `users` | Отфильтровать пользователей с суммарным трафиком ниже порога. | +| `min_total_megabytes` | `users` | Порог суммарного трафика пользователя в MiB (строго больше 0). | +| `min_total_octets` | `users` | Устаревший вариант порога в октетах (если задан `min_total_megabytes`, он приоритетнее). | ## Конфигурация (опционально) @@ -48,5 +51,5 @@ aggregate: 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" +curl -sS "http://127.0.0.1:8080/api/agg/users?include_links=false&min_total_megabytes=1" ``` diff --git a/internal/aggregate/handlers.go b/internal/aggregate/handlers.go index 71c6b4f..a4d2f20 100644 --- a/internal/aggregate/handlers.go +++ b/internal/aggregate/handlers.go @@ -164,8 +164,12 @@ func (h *Handler) handleUsers(w http.ResponseWriter, r *http.Request) { return } includeLinks := strings.EqualFold(r.URL.Query().Get("include_links"), "true") - minOct := uint64(0) - if v := r.URL.Query().Get("min_total_octets"); v != "" { + var minOct uint64 + if v := r.URL.Query().Get("min_total_megabytes"); v != "" { + if n, err := strconv.ParseFloat(v, 64); err == nil && n > 0 { + minOct = uint64(n * float64(mebibyte)) + } + } else if v := r.URL.Query().Get("min_total_octets"); v != "" { if n, err := strconv.ParseUint(v, 10, 64); err == nil { minOct = n } diff --git a/internal/aggregate/handlers_test.go b/internal/aggregate/handlers_test.go index 0b322d5..c7bd625 100644 --- a/internal/aggregate/handlers_test.go +++ b/internal/aggregate/handlers_test.go @@ -49,7 +49,7 @@ func TestHandlerResolveAndFetch(t *testing.T) { if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil { t.Fatal(err) } - if !env.OK || env.Data.FleetTotalOctets != 42 { + if !env.OK || env.Data.FleetTotalMegabytes != octetsToMegabytes(42) { t.Fatalf("data: %+v", env.Data) } } diff --git a/internal/aggregate/merge.go b/internal/aggregate/merge.go index e9ae00a..159be18 100644 --- a/internal/aggregate/merge.go +++ b/internal/aggregate/merge.go @@ -18,7 +18,7 @@ func BuildTraffic(results []ServerFetchResult) []TrafficRow { perUser[u.Username] = map[string]TrafficServerStats{} } perUser[u.Username][fr.Alias] = TrafficServerStats{ - TotalOctets: u.TotalOctets, + TotalMegabytes: octetsToMegabytes(u.TotalOctets), CurrentConnections: u.CurrentConnections, Revision: fr.Revision, } @@ -162,7 +162,7 @@ func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets u m[u.Username] = a } a.byServer[fr.Alias] = TrafficServerStats{ - TotalOctets: u.TotalOctets, + TotalMegabytes: octetsToMegabytes(u.TotalOctets), CurrentConnections: u.CurrentConnections, Revision: fr.Revision, } @@ -192,7 +192,7 @@ func BuildUsers(results []ServerFetchResult, includeLinks bool, minTotalOctets u a := m[name] rows = append(rows, UsersRow{ Username: name, - TotalOctets: a.total, + TotalMegabytes: octetsToMegabytes(a.total), ByServer: a.byServer, Links: a.links, ActiveUniqueIPs: a.act, @@ -212,6 +212,7 @@ func BuildSummary(results []ServerFetchResult, topN int) SummaryData { } sumByUser := map[string]uint64{} + maxUniqueIPByUser := map[string]uint64{} var fleetOctets, fleetConn uint64 ok, fail := 0, 0 serverRows := make([]ServerFetchResult, 0, len(results)) @@ -235,6 +236,9 @@ func BuildSummary(results []ServerFetchResult, topN int) SummaryData { fleetOctets += u.TotalOctets fleetConn += u.CurrentConnections sumByUser[u.Username] += u.TotalOctets + if u.ActiveUniqueIPs > maxUniqueIPByUser[u.Username] { + maxUniqueIPByUser[u.Username] = u.ActiveUniqueIPs + } } } @@ -257,7 +261,25 @@ func BuildSummary(results []ServerFetchResult, topN int) SummaryData { } top := make([]TopUser, 0, len(pairs)) for _, p := range pairs { - top = append(top, TopUser{Username: p.name, TotalOctets: p.n}) + top = append(top, TopUser{Username: p.name, TotalMegabytes: octetsToMegabytes(p.n)}) + } + + ipPairs := make([]pair, 0, len(maxUniqueIPByUser)) + for n, v := range maxUniqueIPByUser { + ipPairs = append(ipPairs, pair{name: n, n: v}) + } + sort.Slice(ipPairs, func(i, j int) bool { + if ipPairs[i].n != ipPairs[j].n { + return ipPairs[i].n > ipPairs[j].n + } + return ipPairs[i].name < ipPairs[j].name + }) + if len(ipPairs) > topN { + ipPairs = ipPairs[:topN] + } + topIP := make([]TopUserByUniqueIPs, 0, len(ipPairs)) + for _, p := range ipPairs { + topIP = append(topIP, TopUserByUniqueIPs{Username: p.name, UniqueIPs: p.n}) } return SummaryData{ @@ -265,8 +287,9 @@ func BuildSummary(results []ServerFetchResult, topN int) SummaryData { ServersTotal: len(results), ServersOK: ok, ServersFailed: fail, - FleetTotalOctets: fleetOctets, + FleetTotalMegabytes: octetsToMegabytes(fleetOctets), FleetTotalConnections: fleetConn, TopUsers: top, + TopUsersByUniqueIPs: topIP, } } diff --git a/internal/aggregate/merge_test.go b/internal/aggregate/merge_test.go index 431da18..c02a6b2 100644 --- a/internal/aggregate/merge_test.go +++ b/internal/aggregate/merge_test.go @@ -35,7 +35,7 @@ func TestBuildTraffic(t *testing.T) { if u1 == nil { t.Fatal("missing u1") } - if u1.Servers["a"].TotalOctets != 100 || u1.Servers["b"].TotalOctets != 30 { + if u1.Servers["a"].TotalMegabytes != octetsToMegabytes(100) || u1.Servers["b"].TotalMegabytes != octetsToMegabytes(30) { t.Fatalf("u1 servers: %+v", u1.Servers) } } @@ -98,15 +98,24 @@ func TestBuildUniqueIPsPrimary(t *testing.T) { func TestBuildSummary(t *testing.T) { results := []ServerFetchResult{ - {Alias: "a", OK: true, Users: []UserInfo{{Username: "x", TotalOctets: 100, CurrentConnections: 2}}}, + { + Alias: "a", OK: true, + Users: []UserInfo{ + {Username: "x", TotalOctets: 100, CurrentConnections: 2, ActiveUniqueIPs: 3}, + {Username: "y", TotalOctets: 500, CurrentConnections: 0, ActiveUniqueIPs: 1}, + }, + }, {Alias: "b", OK: false, Error: "down"}, } s := BuildSummary(results, 5) - if s.ServersOK != 1 || s.ServersFailed != 1 || s.FleetTotalOctets != 100 || s.FleetTotalConnections != 2 { + if s.ServersOK != 1 || s.ServersFailed != 1 || s.FleetTotalMegabytes != octetsToMegabytes(600) || s.FleetTotalConnections != 2 { t.Fatalf("summary: %+v", s) } - if len(s.TopUsers) != 1 || s.TopUsers[0].Username != "x" { - t.Fatalf("top: %+v", s.TopUsers) + if len(s.TopUsers) != 2 || s.TopUsers[0].Username != "y" || s.TopUsers[0].TotalMegabytes != octetsToMegabytes(500) { + t.Fatalf("top traffic: %+v", s.TopUsers) + } + if len(s.TopUsersByUniqueIPs) != 2 || s.TopUsersByUniqueIPs[0].Username != "x" || s.TopUsersByUniqueIPs[0].UniqueIPs != 3 { + t.Fatalf("top unique ips: %+v", s.TopUsersByUniqueIPs) } } diff --git a/internal/aggregate/types.go b/internal/aggregate/types.go index 96173d0..5a9ee31 100644 --- a/internal/aggregate/types.go +++ b/internal/aggregate/types.go @@ -47,11 +47,11 @@ type TrafficRow struct { Servers map[string]TrafficServerStats `json:"servers"` } -// TrafficServerStats per-server counters for one user. +// 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"` + TotalMegabytes float64 `json:"total_megabytes"` + CurrentConnections uint64 `json:"current_connections"` + Revision string `json:"revision,omitempty"` } // UniqueIPsRow describes IP presence across servers for one user. @@ -70,12 +70,12 @@ type IPAssignments struct { // 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"` + Username string `json:"username"` + TotalMegabytes float64 `json:"total_megabytes"` + ByServer map[string]TrafficServerStats `json:"by_server"` + Links *UserLinks `json:"links,omitempty"` + ActiveUniqueIPs uint64 `json:"active_unique_ips"` + RecentUniqueIPs uint64 `json:"recent_unique_ips"` } // SummaryData fleet snapshot. @@ -84,13 +84,20 @@ type SummaryData struct { 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"` + FleetTotalMegabytes float64 `json:"fleet_total_megabytes"` + FleetTotalConnections uint64 `json:"fleet_total_connections"` + TopUsers []TopUser `json:"top_users"` // по суммарному трафику (MiB) + TopUsersByUniqueIPs []TopUserByUniqueIPs `json:"top_users_by_unique_ips"` } -// TopUser by summed total_octets across servers for one username. +// TopUser by summed traffic across servers for one username (мегабайты). type TopUser struct { - Username string `json:"username"` - TotalOctets uint64 `json:"total_octets"` + Username string `json:"username"` + TotalMegabytes float64 `json:"total_megabytes"` +} + +// TopUserByUniqueIPs — max(active_unique_ips) по серверам для username (снимок Telemt). +type TopUserByUniqueIPs struct { + Username string `json:"username"` + UniqueIPs uint64 `json:"unique_ips"` } diff --git a/internal/aggregate/units.go b/internal/aggregate/units.go new file mode 100644 index 0000000..84e04fb --- /dev/null +++ b/internal/aggregate/units.go @@ -0,0 +1,8 @@ +package aggregate + +// mebibyte — двоичный мегабайт (1024² октетов), в JSON полях *megabytes. +const mebibyte = 1024 * 1024 + +func octetsToMegabytes(o uint64) float64 { + return float64(o) / float64(mebibyte) +}