Netpulse_SasS/server/internal/grpcapi/plan_sync.go

package grpcapi

import (
	"context"
	"time"

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

	npv1 "github.com/netpulse/netpulse/gen/go/netpulse/v1"
)

// PlanSyncInterval — як часто звіряти план зонда з базою.
//
// П'ять секунд: людина, яка щойно додала чек у вебі, ще дивиться на
// екран, і затримка в цих межах читається як «спрацювало». Робити це
// частіше немає сенсу — звірка все одно впирається в один SELECT на
// зонда, а сплеск таких запитів на великому тенанті дорожчий за
// секунду очікування.
const PlanSyncInterval = 5 * time.Second

// SyncPlans доганяє зонди змінами, зробленими в UI.
//
// Зміни чеків робить REST API — окремий процес, який не має живої
// сесії зонда. Черги тут не потрібно: план цілком описується своїм
// хешем, тож достатньо періодично звіряти хеш у базі з тим, що зараз
// у зонда, і перезалити план при розбіжності. Пропущений тік нічого не
// ламає — наступний зробить те саме.
func (s *Service) SyncPlans(ctx context.Context) {
	t := time.NewTicker(PlanSyncInterval)
	defer t.Stop()

	id := time.NewTicker(IdentifySyncInterval)
	defer id.Stop()

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

// IdentifySyncInterval — як часто перевіряти, кому бракує розпізнавання.
//
// Окремий, повільніший такт, ніж у звірки планів. Звірка читає хеш,
// а це — запит із записом; ганяти його щоп'ять секунд на кожен зонд
// заради події, яка трапляється при заведенні хоста, не варто. Хвилини
// досить: рівно стільки чекатиме людина, що допіру дала хосту SNMP.
const IdentifySyncInterval = time.Minute

// ensureIdentifyTick заводить чеки розпізнавання новим хостам.
//
// Самої лише появи чека досить: він змінює хеш плану, і найближча
// звірка сама донесе його до зонда.
func (s *Service) ensureIdentifyTick(ctx context.Context) {
	s.mu.RLock()
	sessions := make([]*agentSession, 0, len(s.sessions))
	for _, sess := range s.sessions {
		sessions = append(sessions, sess)
	}
	s.mu.RUnlock()

	for _, sess := range sessions {
		made, err := s.store.EnsureIdentifyChecks(ctx, sess.agent)
		if err != nil {
			s.log.Warn("чек розпізнавання", "agent", sess.agent.ID, "err", err)
			continue
		}
		if made > 0 {
			s.log.Info("заведено чеки розпізнавання",
				"agent", sess.agent.ID, "кількість", made)
		}
	}
}

func (s *Service) syncPlansTick(ctx context.Context) {
	s.mu.RLock()
	sessions := make([]*agentSession, 0, len(s.sessions))
	for _, sess := range s.sessions {
		sessions = append(sessions, sess)
	}
	s.mu.RUnlock()

	for _, sess := range sessions {
		hash, err := s.store.PlanHash(ctx, sess.agent)
		if err != nil {
			s.log.Warn("звірка плану: хеш", "agent", sess.agent.ID, "err", err)
			continue
		}
		if equalBytes(hash, sess.currentPlanHash()) {
			continue
		}

		plan, err := s.store.BuildPlan(ctx, sess.agent)
		if err != nil {
			s.log.Warn("звірка плану: побудова", "agent", sess.agent.ID, "err", err)
			continue
		}

		// Повний план, а не дельта: дельта вміє додавати й міняти, але
		// не знає, що зникло. Порівняння двох хешів теж не знає — воно
		// каже лише «інакше». Перезалив кількох тисяч задач раз на
		// зміну дешевший за чек, який лишився опитувати видалений хост.
		//
		// Модулі перевідсилаються заразом: у плані міг з'явитись
		// перший snmp-чек на зонді, де модуль snmp ще не вмикали.
		s.push(sess, &npv1.ControlDown{
			Payload: &npv1.ControlDown_ModuleControl{
				ModuleControl: store.ModulesForPlan(plan, sess.agent.Modules)},
		})

		// Креденшели — теж заразом, і це не зайва обережність.
		// Знайдено живим прогоном: хост, приписаний зонду вже після
		// його підключення, отримував задачі й падав на кожній із них
		// із «немає SNMP-креденшелів». Пачка видається на Hello, а
		// тоді цього хоста в ній ще не було.
		if bundle, err := s.store.BuildCredentialBundle(ctx, sess.agent, s.ring); err != nil {
			s.log.Warn("звірка плану: креденшели", "agent", sess.agent.ID, "err", err)
		} else if len(bundle.ByDevice) > 0 {
			s.push(sess, &npv1.ControlDown{
				Payload: &npv1.ControlDown_Credentials{Credentials: bundle},
			})
		}

		s.push(sess, &npv1.ControlDown{
			Payload: &npv1.ControlDown_TaskPlan{TaskPlan: plan},
		})

		// Позначку ставимо одразу після постановки в чергу, а не після
		// підтвердження: черга сесії й так упорядкована, а зонд, який
		// відвалився, при повторному Hello принесе свій хеш і отримає
		// план заново.
		sess.setPlanHash(plan.GetPlanHash())

		s.log.Info("план зонда оновлено",
			"agent", sess.agent.ID, "задач", len(plan.GetTasks()))
	}
}

// setAgentPlanHash оновлює позначку після пуша дельти.
//
// Без цього наступний тік звірки побачив би розбіжність і перезалив
// увесь план услід за дельтою, яка вже все зробила.
func (s *Service) setAgentPlanHash(agentID string, hash []byte) {
	s.mu.RLock()
	sess := s.sessions[agentID]
	s.mu.RUnlock()
	if sess != nil {
		sess.setPlanHash(hash)
	}
}