Netpulse_SasS/server/internal/httpapi/queues.go

package httpapi

import (
	"context"
	"errors"
	"net/http"
	"slices"
	"sync"
	"time"

	"github.com/netpulse/netpulse/server/internal/store"
)

// Сторінка «Черги»: чи система встигає, і якщо ні — де саме.
//
// Ціна самої сторінки тут важить не менше за її користь. Сторінка, яку
// відкривають саме тоді, коли системі важко, і яка щодві секунди рахує
// агрегати по гіпертаблицях, добиває базу, яку мала стерегти. Тому:
//
//   - один знімок на всі черги, а не ендпоїнт на кожну;
//   - знімок кешується на кілька секунд і ділиться між усіма, хто
//     дивиться: десять відкритих вкладок мають коштувати як одна;
//   - самі запити читають хвости за частковими індексами, а не
//     перераховують таблиці (див. store/queues.go).
const (
	// Вікно, за яким рахуються надходження, обробка й втрати.
	// Година, а не хвилини: бекап конфігу трапляється раз на добу на
	// хост, і на п'ятнадцятихвилинному вікні здорова система показувала
	// б нулі в усіх стовпчиках.
	queueWindow = time.Hour

	// Скільки живе знімок. П'ять секунд — це менше, ніж такт
	// диспетчера, тож нічого не пропускається, і водночас достатньо,
	// щоб опитування з кількох вкладок злилось в один запит до бази.
	queueTTL = 5 * time.Second

	// Через скільки оновлюється точка відліку для накопичувальних
	// лічильників. Дві хвилини: менше — і приріст стрибав би від
	// одного heartbeat (їх раз на півхвилини), більше — і сторінка
	// довго показувала б «щойно втрачено» після того, як усе минулось.
	queueBaseAge = 2 * time.Minute
)

// queueTenantState — пам'ять спостерігача про один кабінет.
//
// Живе в процесі, а не в базі: потрібна лише для того, щоб відрізнити
// «лічильник ріс щойно» від «лічильник ненульовий з учора». Після
// перезапуску API спостереження чесно починається заново — і сторінка
// про це так і каже, замість вигадати приріст.
type queueTenantState struct {
	report   *store.QueueReport
	reportAt time.Time

	// Знімок, з якого зроблено звіт, і приріст до нього. Потрібні
	// детальному переліку: він мусить пояснювати рівно те число, яке
	// людина щойно бачила на картці. Перезняти їх у момент кліку
	// означало б показати перелік, що суперечить числу над ним.
	facts  *store.QueueFacts
	deltas store.QueueDeltas

	// Деталі кешуються окремо від звіту й окремо по кожній черзі:
	// відкритий перелік теж оновлюється, а платити за нього повним
	// знімком усіх черг немає за що.
	details   map[string]*store.QueueDetail
	detailsAt map[string]time.Time

	baseAt      time.Time
	baseDropped map[string]int64
	basePool    int64
}

type queueWatcher struct {
	mu      sync.Mutex
	tenants map[string]*queueTenantState
}

var queues = &queueWatcher{tenants: map[string]*queueTenantState{}}

// report віддає знімок — свіжий або щойно зроблений.
//
// Збір відбувається під тим самим замком, що й читання кешу, навмисно:
// коли база пригальмовує, десяток одночасних запитів має чекати на один
// збір, а не влаштувати десять власних. Саме в цю мить сторінка й
// небезпечна для бази.
func (w *queueWatcher) report(ctx context.Context, st *store.Store,
	tenantID string) (*store.QueueReport, error) {

	w.mu.Lock()
	defer w.mu.Unlock()

	t := w.tenants[tenantID]
	if t == nil {
		t = &queueTenantState{}
		w.tenants[tenantID] = t
	}
	now := time.Now()
	if t.report != nil && now.Sub(t.reportAt) < queueTTL {
		return t.report, nil
	}

	facts, err := st.CollectQueueFacts(ctx, tenantID, queueWindow)
	if err != nil {
		return nil, err
	}

	d := store.QueueDeltas{AgentDropped: map[string]int64{}}
	if !t.baseAt.IsZero() {
		d.ObservedSec = int(now.Sub(t.baseAt).Seconds())
		for _, a := range facts.Agents {
			if a.Dropped == nil {
				continue
			}
			was, ok := t.baseDropped[a.ID]
			// Лічильник, що поїхав назад, означає перезапуск зонда, а не
			// втрату «мінус сто»: за точку відліку беремо нове значення.
			if ok && *a.Dropped > was {
				d.AgentDropped[a.ID] = *a.Dropped - was
			}
		}
		if facts.Pool.EmptyAcquires > t.basePool {
			d.PoolEmpty = facts.Pool.EmptyAcquires - t.basePool
		}
	}

	if t.baseAt.IsZero() || now.Sub(t.baseAt) >= queueBaseAge {
		t.baseAt = now
		t.basePool = facts.Pool.EmptyAcquires
		t.baseDropped = make(map[string]int64, len(facts.Agents))
		for _, a := range facts.Agents {
			if a.Dropped != nil {
				t.baseDropped[a.ID] = *a.Dropped
			}
		}
	}

	t.report, t.reportAt = store.BuildQueueReport(facts, d), now
	t.facts, t.deltas = facts, d
	// Новий знімок робить старі переліки застарілими: лишити їх означало
	// б показувати під новим числом старі рядки.
	t.details, t.detailsAt = nil, nil
	return t.report, nil
}

// detail віддає перелік за однією чергою.
//
// Знімок для нього не збирається окремо: береться той самий, з якого
// зроблено картку (за потреби — свіжий, через report). Інакше перелік
// пояснював би одне число, а на екрані поруч стояло б інше.
func (w *queueWatcher) detail(ctx context.Context, st *store.Store,
	tenantID, key string) (*store.QueueDetail, error) {

	// Знімок береться поза власним замком — report бере його сам.
	if _, err := w.report(ctx, st, tenantID); err != nil {
		return nil, err
	}

	w.mu.Lock()
	defer w.mu.Unlock()

	t := w.tenants[tenantID]
	if t == nil || t.facts == nil {
		return nil, errNoSnapshot
	}
	now := time.Now()
	if at, ok := t.detailsAt[key]; ok && now.Sub(at) < queueTTL {
		return t.details[key], nil
	}

	det, err := st.QueueDetailFor(ctx, tenantID, key, queueWindow, t.facts, t.deltas)
	if err != nil {
		return nil, err
	}
	if t.details == nil {
		t.details, t.detailsAt = map[string]*store.QueueDetail{}, map[string]time.Time{}
	}
	t.details[key], t.detailsAt[key] = det, now
	return det, nil
}

var errNoSnapshot = errors.New("знімка черг ще немає")

// handleQueues — GET /api/v1/queues.
//
// Право agents:read, а не власне: сторінка не показує нічого, чого не
// показують «Зонди», — вона лише складає з цього висновок. Заводити під
// висновок окреме право означало б, що людина, яка відповідає за збір,
// не бачить, чи він устигає.
func (s *Server) handleQueues(w http.ResponseWriter, r *http.Request, p *Principal) {
	if !p.Can("agents:read") {
		writeError(w, http.StatusForbidden, "forbidden", "немає права agents:read")
		return
	}

	rep, err := queues.report(r.Context(), s.store, p.TenantID)
	if err != nil {
		s.log.Error("знімок черг", "err", err)
		writeError(w, http.StatusInternalServerError, "internal", "внутрішня помилка")
		return
	}
	writeJSON(w, http.StatusOK, rep)
}

// handleQueueDetail — GET /api/v1/queues/{key}.
//
// Окремий виклик, а не поле у знімку, навмисно. Знімок їде раз на п'ять
// секунд усім, хто відкрив сторінку; перелік зачеплених хостів потрібен
// одному й лише після того, як він натиснув на картку. Возити його
// разом зі знімком означало б платити за нього постійно, а дивитись —
// зрідка.
func (s *Server) handleQueueDetail(w http.ResponseWriter, r *http.Request, p *Principal) {
	if !p.Can("agents:read") {
		writeError(w, http.StatusForbidden, "forbidden", "немає права agents:read")
		return
	}

	key := r.PathValue("key")
	// Білий список, а не довільний ключ у switch: інакше друкарська
	// помилка в адресі давала б 500 замість зрозумілої відмови.
	if !slices.Contains(store.QueueDetailKeys(), key) {
		writeError(w, http.StatusNotFound, "not_found", "немає такої черги: "+key)
		return
	}

	det, err := queues.detail(r.Context(), s.store, p.TenantID, key)
	if err != nil {
		s.log.Error("деталі черги", "key", key, "err", err)
		writeError(w, http.StatusInternalServerError, "internal", "внутрішня помилка")
		return
	}
	writeJSON(w, http.StatusOK, det)
}