Netpulse_SasS/server/internal/grpcapi/storage_retention.go

package grpcapi

import (
	"context"
	"time"
)

// dataRetentionLockKey — окреме advisory-блокування під прибирання
// даних.
//
// Своє, а не спільне з очисткою конфігів (retentionLockKey): обидва
// такти тікають незалежно, і спільний ключ означав би, що довгий перший
// прохід по прогонах команд заодно зупиняє прибирання версій конфігів.
const dataRetentionLockKey = 0x6e70_6472 // "npdr"

// DataRetentionInterval — як часто прибирати й міряти.
//
// Година. Найдрібніша одиниця строку — доба, тож частіше шукати нема
// чого. Рідше — означало б, що інсталяція, яка вперше поставила строк,
// починає звільняти місце невідомо коли; а на переповненому диску
// «невідомо коли» — це вже після аварії.
const DataRetentionInterval = time.Hour

// SweepDataRetention прибирає застарілі дані за налаштованими строками
// й веде ряд спостережень за розміром.
//
// Живе в колекторі, а не в REST-процесі, з тих самих трьох міркувань,
// що й очистка конфігів (ncm_retention.go): тут стоїть той, хто дані
// СТВОРЮЄ; REST-процес масштабують копіями й він же обслуговує запити
// людини; ключ шифрування тут не потрібен зовсім.
//
// Кілька екземплярів колектора безпечні: тік бере advisory-блокування,
// тож працює рівно один.
func (s *Service) SweepDataRetention(ctx context.Context) {
	t := time.NewTicker(DataRetentionInterval)
	defer t.Stop()

	s.log.Info("прибиральник даних запущено", "інтервал", DataRetentionInterval)

	// Перший знімок розміру знімається одразу, не чекаючи години.
	// Приріст рахується з РІЗНИЦІ спостережень, тож перше з них має
	// з'явитись якомога раніше: до нього сторінка не має що сказати про
	// швидкість, а саме заради швидкості її й відкривають.
	s.dataRetentionTick(ctx)

	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			s.dataRetentionTick(ctx)
		}
	}
}

func (s *Service) dataRetentionTick(ctx context.Context) {
	conn, err := s.store.Pool().Acquire(ctx)
	if err != nil {
		s.log.Error("прибиральник даних: з'єднання", "err", err)
		return
	}
	defer conn.Release()

	var got bool
	if err := conn.QueryRow(ctx, `SELECT pg_try_advisory_lock($1)`,
		int64(dataRetentionLockKey)).Scan(&got); err != nil {
		s.log.Error("прибиральник даних: блокування", "err", err)
		return
	}
	if !got {
		// Прибирає інший інстанс — штатний стан.
		return
	}
	defer func() {
		_, _ = conn.Exec(context.WithoutCancel(ctx),
			`SELECT pg_advisory_unlock($1)`, int64(dataRetentionLockKey))
	}()

	// 1. Звірка політик TimescaleDB зі збереженими строками.
	//
	// Політики накладаються одразу при збереженні, тож у нормальному
	// житті тут завжди нуль. Такт потрібен для іншого: політику знімають
	// руками під час обслуговування бази, а pg_dump не везе фонових
	// задач TimescaleDB узагалі — тобто після відновлення з дампа
	// строки в таблиці є, а видалення не працює, і дізнаються про це
	// лише тоді, коли диск уже закінчився. Один запит на годину — мала
	// ціна за те, щоб цей стан не пережив першої ж години.
	if n, err := s.store.SyncRetentionPolicies(ctx); err != nil {
		s.log.Error("прибиральник даних: політики TimescaleDB", "err", err)
	} else if n > 0 {
		s.log.Warn("політики зберігання розійшлися зі строками й виправлені",
			"політик", n)
	}

	// 2. Пакетне прибирання звичайних таблиць.
	stats, err := s.store.SweepDataRetention(ctx)
	if err != nil {
		s.log.Error("прибиральник даних: прохід", "err", err)
	}
	for _, st := range stats {
		if st.Deleted == 0 {
			// Підпадало, а не прибралось. Помилки немає, і саме тому це
			// найнебезпечніший зі станів: тиша читається як «усе
			// гаразд». Майже завжди причина одна — з'єднання відкрито
			// роллю під RLS без заданого кабінету, тобто NETPULSE_DSN
			// колектора не веде на роль netpulse_worker (0063).
			s.log.Warn("прибирання не зачепило нічого, хоча дані під строком є",
				"вид", st.Kind, "підпадало", st.Due,
				"ймовірна_причина", "з'єднання під RLS без BYPASSRLS — перевірте DSN колектора")
			continue
		}
		s.log.Info("прибрано застарілі дані",
			"вид", st.Kind, "рядків", st.Deleted, "лишилось_ще", st.More)
	}

	// 3. Знімок розмірів.
	//
	// Після прибирання, а не до: інакше ряд спостережень описував би
	// базу, якої вже немає, і приріст за добу вважався б із піків перед
	// кожним проходом.
	if err := s.store.SampleStorage(ctx); err != nil {
		s.log.Error("прибиральник даних: знімок розмірів", "err", err)
	}

	// 4. Попередження про вичерпання місця.
	//
	// Тут, а не в движку правил, і не окремим тактом.
	//
	// Не в движку — бо метричне правило обчислюється запитом із JOIN
	// inv.devices, тобто вимагає хоста, а хоста для тому бази не існує
	// (повний розбір — у 0067_storage_alert.sql і store/storage_alert.go).
	//
	// Не окремим тактом — бо перевірка спирається рівно на ті числа, які
	// щойно ліг у ряд знімок вище. Свій такт означав би, що алерт іноді
	// рахується з даних, старших за годину, і найчастіше — саме тоді,
	// коли швидкість щойно змінилась.
	//
	// Одразу ПІСЛЯ знімка, а не до: інакше перша година життя інсталяції
	// проходила б без жодного спостереження, тобто перевірка на першому
	// такті була б сліпою за побудовою.
	if st, err := s.store.CheckStorageAlert(ctx); err != nil {
		s.log.Error("прибиральник даних: попередження про місце", "err", err)
	} else if st.Level != "" {
		s.log.Warn("місце на томі бази закінчується",
			"рівень", st.Level, "причина", st.Reason, "текст", st.Message)
	} else if st.Blind != "" && st.Enabled {
		// Найнебезпечніший зі станів і тому не Debug: попередження
		// увімкнене, але спертись йому нема на що. Ззовні це
		// невідрізнимо від «усе гаразд» — рівно те, чого ця робота мала
		// позбутись.
		s.log.Warn("попередження про місце не має на що спертись",
			"причина", st.Blind,
			"що_зробити", "вказати ємність тому або змонтувати том бази в збірник")
	}
}