Netpulse_SasS/server/internal/store/metrics.go

package store

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
)

// SeriesInfo — один ряд метрики, який можна намалювати.
//
// Ключа метрики недостатньо, щоб назвати ряд: на одному хості
// `if.in_bps` існує стільки разів, скільки в нього портів. Тому назва
// збирається з ключа плюс порт або мітки.
type SeriesInfo struct {
	ID            int64             `json:"id"`
	MetricKey     string            `json:"metric_key"`
	Unit          string            `json:"unit,omitempty"`
	InterfaceID   string            `json:"interface_id,omitempty"`
	InterfaceName string            `json:"interface_name,omitempty"`
	Labels        map[string]string `json:"labels,omitempty"`
	LastValue     *float64          `json:"last_value,omitempty"`
	LastAt        *time.Time        `json:"last_at,omitempty"`
}

// ListSeries — які метрики взагалі є в цього хоста.
//
// Останнє значення береться тут же: список метрик без чисел змушував би
// клацнути кожну, щоб дізнатись, чи в ній щось є.
func (s *Store) ListSeries(ctx context.Context, tenantID, deviceID string) ([]SeriesInfo, error) {
	out := []SeriesInfo{}
	err := s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		rows, err := tx.Query(ctx, `
			SELECT s.id, s.metric_key, COALESCE(s.unit,''),
			       COALESCE(s.interface_id::text,''), COALESCE(i.name,''),
			       s.labels::text,
			       (SELECT value FROM ts.samples m
			         WHERE m.series_id = s.id ORDER BY m.ts DESC LIMIT 1),
			       (SELECT ts FROM ts.samples m
			         WHERE m.series_id = s.id ORDER BY m.ts DESC LIMIT 1)
			FROM ts.series s
			LEFT JOIN inv.interfaces i ON i.id = s.interface_id
			WHERE s.tenant_id = $1 AND s.device_id = $2
			ORDER BY s.metric_key, i.name NULLS FIRST
		`, tenantID, deviceID)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var si SeriesInfo
			var labels string
			if err := rows.Scan(&si.ID, &si.MetricKey, &si.Unit,
				&si.InterfaceID, &si.InterfaceName, &labels,
				&si.LastValue, &si.LastAt); err != nil {
				return err
			}
			si.Labels = decodeLabels(labels)
			out = append(out, si)
		}
		return rows.Err()
	})
	return out, err
}

// Point — одна точка графіка.
//
// Val — вказівник, бо пропуск і нуль — різні речі. Лінія, проведена
// через діру в даних, каже «все було добре», хоча насправді нічого не
// відомо; це найгірший різновид брехні в моніторингу.
type Point struct {
	Ts  time.Time `json:"ts"`
	Val *float64  `json:"v"`
	Min *float64  `json:"min,omitempty"`
	Max *float64  `json:"max,omitempty"`
}

// MetricSeries — ряд разом із точками.
type MetricSeries struct {
	ID            int64             `json:"id"`
	MetricKey     string            `json:"metric_key"`
	Unit          string            `json:"unit,omitempty"`
	InterfaceName string            `json:"interface_name,omitempty"`
	Labels        map[string]string `json:"labels,omitempty"`
	Points        []Point           `json:"points"`
}

// MetricQuery — що саме малюємо.
type MetricQuery struct {
	DeviceID  string
	SeriesIDs []int64
	From      time.Time
	To        time.Time
	// Скільки точок хоче намалювати клієнт. Крок рахується з нього, а
	// не приходить ззовні: інакше вузьке вікно з дрібним кроком
	// повернуло б десятки тисяч точок, з яких екран покаже сотні.
	Points int
}

// MetricResult — відповідь на запит графіка.
type MetricResult struct {
	From    time.Time      `json:"from"`
	To      time.Time      `json:"to"`
	StepSec int            `json:"step_sec"`
	Source  string         `json:"source"`
	Series  []MetricSeries `json:"series"`
}

const (
	maxPoints     = 2000
	defaultPoints = 300
)

// pickSource обирає найгрубіший роллап, крок якого ще вміщується у
// потрібний.
//
// Читати сирі дані за місяць — це мільйони рядків заради трьохсот
// пікселів. Але й брати годинні бакети на вікні в п'ять хвилин не
// можна: графік вийде з однієї точки.
func pickSource(step time.Duration) (table string, bucket time.Duration, name string) {
	switch {
	case step < 5*time.Minute:
		return "", 0, "raw"
	case step < time.Hour:
		return "ts.samples_5m", 5 * time.Minute, "5m"
	default:
		return "ts.samples_1h", time.Hour, "1h"
	}
}

// QueryMetrics віддає точки для графіка.
//
// Бакетизація робиться в БД (time_bucket), а не в Go: тягнути сирі
// точки, щоб усереднити їх у застосунку, означає передати по мережі те,
// що однаково буде викинуто.
func (s *Store) QueryMetrics(ctx context.Context, tenantID string, q MetricQuery) (*MetricResult, error) {
	if q.To.IsZero() {
		q.To = time.Now()
	}
	if q.From.IsZero() || !q.From.Before(q.To) {
		q.From = q.To.Add(-6 * time.Hour)
	}
	if q.Points <= 0 {
		q.Points = defaultPoints
	}
	if q.Points > maxPoints {
		q.Points = maxPoints
	}
	if len(q.SeriesIDs) == 0 {
		return &MetricResult{From: q.From, To: q.To, Series: []MetricSeries{}}, nil
	}

	span := q.To.Sub(q.From)
	step := span / time.Duration(q.Points)
	table, bucket, source := pickSource(step)

	// Крок не може бути дрібнішим за бакет джерела: просити 10-секундні
	// точки з п'ятихвилинного роллапу безглуздо — вийде та сама
	// п'ятихвилинка, лише повторена тридцять разів.
	if bucket > 0 && step < bucket {
		step = bucket
	}
	if step < time.Second {
		step = time.Second
	}

	var sql string
	if table == "" {
		sql = `
			SELECT s.series_id,
			       time_bucket($1::interval, s.ts) AS bucket,
			       avg(s.value), min(s.value), max(s.value)
			FROM ts.samples s
			WHERE s.series_id = ANY($2::bigint[]) AND s.ts >= $3 AND s.ts <= $4
			GROUP BY 1, 2 ORDER BY 1, 2`
	} else {
		sql = fmt.Sprintf(`
			SELECT s.series_id,
			       time_bucket($1::interval, s.bucket) AS bucket,
			       avg(s.avg_value), min(s.min_value), max(s.max_value)
			FROM %s s
			WHERE s.series_id = ANY($2::bigint[]) AND s.bucket >= $3 AND s.bucket <= $4
			GROUP BY 1, 2 ORDER BY 1, 2`, table)
	}

	res := &MetricResult{
		From:    q.From,
		To:      q.To,
		StepSec: int(step / time.Second),
		Source:  source,
		Series:  []MetricSeries{},
	}

	err := s.InTenantTx(ctx, tenantID, func(tx pgx.Tx) error {
		// Метадані рядів читаємо під RLS і з явною перевіркою хоста:
		// series_id приходить від клієнта, і без цієї перевірки чужий
		// ідентифікатор віддав би чужі дані. Самі ts.* під RLS не
		// стоять (несумісно зі стисненням), тож фільтр тут — єдиний.
		meta := map[int64]*MetricSeries{}
		order := []int64{}
		rows, err := tx.Query(ctx, `
			SELECT s.id, s.metric_key, COALESCE(s.unit,''),
			       COALESCE(i.name,''), s.labels::text
			FROM ts.series s
			LEFT JOIN inv.interfaces i ON i.id = s.interface_id
			WHERE s.tenant_id = $1 AND s.device_id = $2 AND s.id = ANY($3::bigint[])
			ORDER BY s.metric_key
		`, tenantID, q.DeviceID, q.SeriesIDs)
		if err != nil {
			return err
		}
		for rows.Next() {
			var ms MetricSeries
			var labels string
			if err := rows.Scan(&ms.ID, &ms.MetricKey, &ms.Unit,
				&ms.InterfaceName, &labels); err != nil {
				rows.Close()
				return err
			}
			ms.Labels = decodeLabels(labels)
			ms.Points = []Point{}
			meta[ms.ID] = &ms
			order = append(order, ms.ID)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return err
		}
		if len(order) == 0 {
			return nil
		}

		// Питаємо лише ті ряди, що пройшли перевірку належності.
		prows, err := tx.Query(ctx, sql, step, order, q.From, q.To)
		if err != nil {
			return err
		}
		defer prows.Close()
		for prows.Next() {
			var id int64
			var p Point
			if err := prows.Scan(&id, &p.Ts, &p.Val, &p.Min, &p.Max); err != nil {
				return err
			}
			if m := meta[id]; m != nil {
				m.Points = append(m.Points, p)
			}
		}
		if err := prows.Err(); err != nil {
			return err
		}

		for _, id := range order {
			res.Series = append(res.Series, *meta[id])
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return res, nil
}

// decodeLabels розбирає jsonb-мітки в мапу.
//
// Помилка розбору не має валити графік: мітки — це підпис до лінії, а
// не самі дані. Порожня мапа гірша за жодну лише на вигляд.
func decodeLabels(raw string) map[string]string {
	if raw == "" || raw == "{}" {
		return nil
	}
	var m map[string]string
	if err := json.Unmarshal([]byte(raw), &m); err != nil {
		return nil
	}
	if len(m) == 0 {
		return nil
	}
	return m
}