package ui // Вкладка «Монитор» (порт awg-ui/monitor.ts): встроенный Prometheus и основные графики. // GET /ui/monitor — готов ли Prometheus, можно ли поставить его кнопкой, ход установки // POST /ui/monitor/install — попросить awg-ctrl выполнить `awg-ctrl monitor install` (root) // POST /ui/monitor/remove — то же для `awg-ctrl monitor remove` // GET /ui/monitor/overview — серии для графиков за 1 ч / 6 ч / 24 ч / 7 дн / 30 дн // Панель сама пакетов не ставит и в Prometheus ходит только фиксированными запросами. // Prometheus — на этой машине или на ноде (панель без VPN, например после переноса): тогда и // запросы к нему, и его скрейп метрик панели идут через агента ноды по её WSS. import ( "errors" "io" "math" "net/http" "net/url" "sort" "strconv" "strings" "sync" "time" "github.com/maeneko/forgetting/internal/jsjson" ) type promRange struct{ seconds, step int64 } // Шаг — чтобы на графике было ~250–360 точек; окно rate — минимум два скрейпа (15 с). // Дальше 30 дней не заглядываем: столько Prometheus хранит (promRetention в internal/cli/monitor.go). var promRanges = map[string]promRange{ "1h": {3600, 15}, "6h": {21600, 60}, "24h": {86400, 300}, "7d": {604800, 1800}, "30d": {2592000, 7200}, } // Окно rate() — не короче шага: иначе на неделе/месяце точка была бы случайной двухминутной выборкой // из двух часов, а не средним за них. Значения «сейчас» при этом считаются отдельно, по окну 2m. const promRateWindow = "2m" func withWindow(q, w string) string { return strings.ReplaceAll(q, "["+promRateWindow+"]", "["+w+"]") } // rated — ряды, чьё «сейчас» показывает вкладка; при расширенном окне их берём мгновенным запросом. var rated = []string{"down", "up", "cpu"} // Запросы по серверам. Трафик — со счётчиков интерфейса (монотонны), а не с сумм по пирам. var promQueries = []struct{ key, q string }{ {"down", `sum by (server_id, server) (rate(awg_link_transmit_bytes_total[2m])) * 8`}, {"up", `sum by (server_id, server) (rate(awg_link_receive_bytes_total[2m])) * 8`}, {"online", `sum by (server_id, server) (awg_peers_online)`}, {"heavy", `sum by (server_id, server) (awg_shaping_heavy_devices)`}, {"drops", `sum by (server_id, server) (rate(awg_shaping_qdisc_drops_total[2m]))`}, {"congested", `max by (server_id, server) (awg_shaping_congested)`}, {"scrape_ok", `max by (server_id, server) (awg_scrape_ok)`}, {"cpu", `100 * (1 - sum by (server_id, server) (rate(awg_host_cpu_seconds_total{mode=~"idle|iowait"}[2m])) / sum by (server_id, server) (rate(awg_host_cpu_seconds_total[2m])))`}, {"mem", `100 * (1 - sum by (server_id, server) (awg_host_memory_available_bytes) / sum by (server_id, server) (awg_host_memory_total_bytes))`}, {"cores", `max by (server_id, server) (awg_host_cpu_cores)`}, {"mem_total", `max by (server_id, server) (awg_host_memory_total_bytes)`}, } type point [2]float64 type seriesOne struct { name any points []point } // series — server_id → ряд, в порядке ответа Prometheus (как Map в JS). type series struct { ids []int m map[int]*seriesOne } func (s *series) get(id int) *seriesOne { if s == nil { return nil } return s.m[id] } type Monitor struct { a *App promURL string localEnabled bool http *http.Client srcMu sync.Mutex src *promSource // найденный Prometheus; nil — ещё не искали или не нашли srcAt time.Time } // promSource — где Prometheus: local — на этой машине, иначе на ноде node (через её агента). type promSource struct { local bool node int } // Найденный Prometheus помним недолго; ненайденный ищем при каждом запросе (вкладка // запрашивает статус редко, а во время установки должна увидеть его сразу). const promSourceTTL = 30 * time.Second func newMonitor(a *App) *Monitor { return &Monitor{ a: a, promURL: envOr("PROM_URL", "http://127.0.0.1:9090"), localEnabled: strings.ToLower(envOr("LOCAL_NODE", "on")) != "off", http: &http.Client{Transport: &http.Transport{Proxy: nil}}, } } var errProm = errors.New("Prometheus не отвечает") // source — свой Prometheus, если он есть; иначе первый по id сервер, чей агент отдаёт свой. func (m *Monitor) source() *promSource { m.srcMu.Lock() if m.src != nil && time.Since(m.srcAt) < promSourceTTL { s := m.src m.srcMu.Unlock() return s } m.srcMu.Unlock() s := m.findSource() m.srcMu.Lock() m.src, m.srcAt = s, time.Now() m.srcMu.Unlock() return s } // forgetSource — Prometheus поставили, удалили или он перестал отвечать: искать заново. func (m *Monitor) forgetSource() { m.srcMu.Lock() m.src = nil m.srcMu.Unlock() } func (m *Monitor) findSource() *promSource { if m.promReady() { return &promSource{local: true} } ids := m.a.nodes.IDs() ready := make([]bool, len(ids)) var wg sync.WaitGroup for i, id := range ids { // Сервер 0 со своим awg-ctrl — эта же машина: её Prometheus уже проверен выше. if id == 0 && m.localEnabled || !m.a.nodes.IsOnline(id) { continue } wg.Add(1) go func() { defer wg.Done() r, err := m.a.nodes.CtrlFor(id)("GET", "/prom/-/ready", nil) ready[i] = err == nil && r.Status == 200 }() } wg.Wait() for i, id := range ids { if ready[i] { return &promSource{node: id} } } return nil } // promGet — JSON-ответ Prometheus (path с запросом): напрямую или через агента ноды. func (m *Monitor) promGet(src *promSource, path string, q url.Values) (any, error) { if src == nil { return nil, errProm } path += "?" + q.Encode() if !src.local { r, err := m.a.nodes.CtrlFor(src.node)("GET", "/prom"+path, nil) if err != nil || r.Status < 200 || r.Status >= 300 { return nil, errProm } return r.Data, nil } c := &http.Client{Transport: m.http.Transport, Timeout: 8 * time.Second} resp, err := c.Get(m.promURL + path) if err != nil { return nil, errProm } defer resp.Body.Close() raw, err := io.ReadAll(resp.Body) if err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300 { return nil, errProm } return decodeLikeAxios(raw), nil } func (m *Monitor) rangeQuery(src *promSource, query string, start, end, step int64) (*series, error) { q := url.Values{} q.Set("query", query) q.Set("start", strconv.FormatInt(start, 10)) q.Set("end", strconv.FormatInt(end, 10)) q.Set("step", strconv.FormatInt(step, 10)) data, err := m.promGet(src, "/api/v1/query_range", q) if err != nil { return nil, err } out := &series{m: map[int]*seriesOne{}} for _, r := range arr(field(data, "data", "result")) { metric := field(r, "metric") idf := jsToNumberAny(field(metric, "server_id")) if field(metric, "server_id") == nil { idf = math.NaN() } if idf != float64(int(idf)) { continue } id := int(idf) name := field(metric, "server") if name == nil { name = strconv.Itoa(id) } s := &seriesOne{name: name} for _, v := range arr(field(r, "values")) { pair := arr(v) if len(pair) < 2 { continue } t := jsToNumberAny(pair[0]) val := jsToNumberAny(pair[1]) // NaN/Inf из деления на ноль отбрасываем: на графике это дыра, а не нулевая скорость if math.IsNaN(val) || math.IsInf(val, 0) { continue } s.points = append(s.points, point{t, val}) } if _, dup := out.m[id]; !dup { out.ids = append(out.ids, id) } out.m[id] = s } return out, nil } // instantQuery — значение на момент time по каждому server_id. func (m *Monitor) instantQuery(src *promSource, query string, time_ int64) (map[int]float64, error) { q := url.Values{} q.Set("query", query) q.Set("time", strconv.FormatInt(time_, 10)) data, err := m.promGet(src, "/api/v1/query", q) if err != nil { return nil, err } out := map[int]float64{} for _, r := range arr(field(data, "data", "result")) { metric := field(r, "metric") if field(metric, "server_id") == nil { continue } idf := jsToNumberAny(field(metric, "server_id")) pair := arr(field(r, "value")) if idf != float64(int(idf)) || len(pair) < 2 { continue } if v := jsToNumberAny(pair[1]); !math.IsNaN(v) && !math.IsInf(v, 0) { out[int(idf)] = v } } return out, nil } // total — сумма (или максимум) рядов по меткам времени. func total(s *series, max bool) []any { acc := map[float64]float64{} var order []float64 for _, id := range s.ids { for _, p := range s.m[id].points { cur, seen := acc[p[0]] if !seen { order = append(order, p[0]) } if max { acc[p[0]] = math.Max(cur, p[1]) } else { acc[p[0]] = cur + p[1] } } } sort.Float64s(order) out := []any{} for _, t := range order { out = append(out, []any{t, acc[t]}) } return out } func last(s *seriesOne) any { if s == nil || len(s.points) == 0 { return nil } return s.points[len(s.points)-1][1] } func (m *Monitor) promReady() bool { c := &http.Client{Transport: m.http.Transport, Timeout: 1500 * time.Millisecond} resp, err := c.Get(m.promURL + "/-/ready") if err != nil { return false } resp.Body.Close() return resp.StatusCode == 200 } type monitorJob struct { can bool action any running bool exit any log string } // target — чей awg-ctrl ставит и удаляет Prometheus: того сервера, где он уже стоит; если его // нет — свой (панель + VPN) или выбранный в сервер-баре (панель без VPN). -1 — некому: например, // Prometheus на машине панели без VPN поставлен руками. func (m *Monitor) target(src *promSource, qs url.Values) int { switch { case src != nil && !src.local: return src.node case m.localEnabled: return 0 case src != nil: return -1 } if f := jsToNumber(strings.Join(qs["server"], ",")); f == float64(int(f)) && f >= 0 && m.a.nodes.Exists(int(f)) { return int(f) } return -1 } // host — сервер target, если это не машина панели (её вкладка так и называет: «этот сервер»). func (m *Monitor) host(target int) any { if target < 0 || target == 0 && m.localEnabled { return nil } for _, s := range m.a.nodes.List() { if s.ID == target { return jsjson.Object{{Key: "id", Val: s.ID}, {Key: "name", Val: s.Name}} } } return nil } // jobState — установка возможна, только если у target есть awg-ctrl и он знает маршрут. func (m *Monitor) jobState(target int) monitorJob { if target < 0 { return monitorJob{} } r, err := m.a.nodes.CtrlFor(target)("GET", "/api/monitor/job", nil) if err != nil || r.Status != 200 { return monitorJob{} } j := monitorJob{can: true, action: field(r.Data, "action"), running: truthy(field(r.Data, "running")), exit: field(r.Data, "exit")} switch l := field(r.Data, "log").(type) { case nil: case string: j.log = l default: j.log, _ = jsString(l) } return j } func (m *Monitor) router() *Router { rt := newRouter() rt.Get("/", func(c *Ctx) bool { src := m.source() ready := src != nil target := m.target(src, c.R.URL.Query()) job := m.jobState(target) var jobOut any if job.can && truthy(job.action) { jobOut = jsjson.Object{{Key: "action", Val: job.action}, {Key: "running", Val: job.running}, {Key: "exit", Val: job.exit}, {Key: "log", Val: job.log}} } return c.JSON(200, jsjson.Object{{Key: "ready", Val: ready}, {Key: "can_install", Val: job.can}, {Key: "job", Val: jobOut}, {Key: "command", Val: "awg-ctrl monitor install"}, {Key: "host", Val: m.host(target)}}) }) // Установка и удаление — одно и то же: просим awg-ctrl сервера, он запускает CLI от root. ask := func(action string) handler { return func(c *Ctx) bool { target := m.target(m.source(), c.R.URL.Query()) if target < 0 { return c.Error(409, "Некому поставить Prometheus: выберите VPN-сервер или выполните команду на машине") } m.forgetSource() r, err := m.a.nodes.CtrlFor(target)("POST", "/api/monitor/"+action, nil) if err != nil { return c.Error(502, "awg-ctrl недоступен") } if r.Status == 404 { return c.Error(501, "awg-ctrl старой версии не умеет этого: обновите сервер или выполните команду вручную") } return c.JSON(r.Status, r.Data) } } rt.Post("/install", ask("install")) rt.Post("/remove", ask("remove")) rt.Get("/overview", func(c *Ctx) bool { qs := c.R.URL.Query() rangeKey := "1h" if _, ok := qs["range"]; ok { rangeKey = strings.Join(qs["range"], ",") } rg, ok := promRanges[rangeKey] if !ok { return c.Error(400, "range: 1h, 6h, 24h, 7d или 30d") } src := m.source() end := now() start := end - rg.seconds window := promRateWindow if rg.step > 120 { window = strconv.FormatInt(rg.step, 10) + "s" } got := make([]*series, len(promQueries)) errs := make([]error, len(promQueries)+2+len(rated)) // Объём за весь период: increase() по тем же монотонным счётчикам (сбросы счётчика он учитывает). // Окно расширено — «сейчас» с конца ряда было бы средним за шаг; берём мгновенные значения по 2m. var usedDown, usedUp map[int]float64 nowVals := map[string]map[int]float64{} var mu sync.Mutex var wg sync.WaitGroup for i, pq := range promQueries { wg.Add(1) go func() { defer wg.Done() got[i], errs[i] = m.rangeQuery(src, withWindow(pq.q, window), start, end, rg.step) }() } span := strconv.FormatInt(rg.seconds, 10) + "s" n := len(promQueries) wg.Add(2) go func() { defer wg.Done() usedDown, errs[n] = m.instantQuery(src, `sum by (server_id, server) (increase(awg_link_transmit_bytes_total[`+span+`]))`, end) }() go func() { defer wg.Done() usedUp, errs[n+1] = m.instantQuery(src, `sum by (server_id, server) (increase(awg_link_receive_bytes_total[`+span+`]))`, end) }() if window != promRateWindow { for j, k := range rated { for _, pq := range promQueries { if pq.key != k { continue } wg.Add(1) go func() { defer wg.Done() v, err := m.instantQuery(src, pq.q, end) errs[n+2+j] = err mu.Lock() nowVals[k] = v mu.Unlock() }() } } } wg.Wait() for _, err := range errs { if err != nil { m.forgetSource() return c.Error(502, err.Error()) } } used := func(v map[int]float64, id int) any { if x, ok := v[id]; ok { return x } return nil } // Вкладка показывает один сервер — выбранный в сервер-баре (?server=). if _, has := qs["server"]; has { if f := jsToNumber(strings.Join(qs["server"], ",")); f == float64(int(f)) { only := int(f) for i, s := range got { f := &series{m: map[int]*seriesOne{}} if x, ok := s.m[only]; ok { f.ids, f.m[only] = []int{only}, x } got[i] = f } } } s := map[string]*series{} for i, pq := range promQueries { s[pq.key] = got[i] } cur := func(k string, id int) any { if v, ok := nowVals[k]; ok { if x, ok := v[id]; ok { return x } return nil } return last(s[k].get(id)) } ids := map[int]bool{} for _, x := range got { for _, id := range x.ids { ids[id] = true } } var sorted []int for id := range ids { sorted = append(sorted, id) } sort.Ints(sorted) servers := []any{} for _, id := range sorted { var name any = strconv.Itoa(id) for _, k := range []string{"scrape_ok", "down", "online"} { if x := s[k].get(id); x != nil { name = x.name break } } servers = append(servers, jsjson.Object{ {Key: "id", Val: id}, {Key: "name", Val: name}, {Key: "ok", Val: last(s["scrape_ok"].get(id)) == 1.0}, {Key: "down", Val: cur("down", id)}, {Key: "up", Val: cur("up", id)}, {Key: "online", Val: last(s["online"].get(id))}, {Key: "heavy", Val: last(s["heavy"].get(id))}, {Key: "congested", Val: last(s["congested"].get(id)) == 1.0}, {Key: "cpu", Val: cur("cpu", id)}, {Key: "mem", Val: last(s["mem"].get(id))}, {Key: "cores", Val: last(s["cores"].get(id))}, {Key: "mem_total", Val: last(s["mem_total"].get(id))}, {Key: "used_down", Val: used(usedDown, id)}, {Key: "used_up", Val: used(usedUp, id)}, }) } var rangeOut any = "1h" if _, ok := qs["range"]; ok { rangeOut = rangeKey } return c.JSON(200, jsjson.Object{ {Key: "range", Val: rangeOut}, {Key: "step", Val: rg.step}, {Key: "from", Val: start}, {Key: "to", Val: end}, {Key: "servers", Val: servers}, {Key: "total", Val: jsjson.Object{ {Key: "down", Val: total(s["down"], false)}, {Key: "up", Val: total(s["up"], false)}, {Key: "online", Val: total(s["online"], false)}, {Key: "heavy", Val: total(s["heavy"], false)}, {Key: "drops", Val: total(s["drops"], false)}, {Key: "cpu", Val: total(s["cpu"], true)}, {Key: "mem", Val: total(s["mem"], true)}, }}, }) }) return rt }