Netpulse_SasS/server/internal/grpcapi/ncm_dispatch.go

package grpcapi

import (
	"context"
	"time"

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

// DispatchConfigJobs роздає завдання збору конфігу живим сесіям.
//
// Опитування таблиці, а не сповіщення: чергу наповнює інший процес
// (REST), і LISTEN/NOTIFY тут дав би доставку «здебільшого» — воно не
// переживає перезапуск слухача. Ціна опитування — один дешевий запит за
// індексом раз на кілька секунд, і лише тоді, коли хоч один зонд
// на зв'язку.
func (s *Service) DispatchConfigJobs(ctx context.Context, every time.Duration) {
	if every <= 0 {
		every = 5 * time.Second
	}
	t := time.NewTicker(every)
	defer t.Stop()

	// Зависле завдання повертається у відмову за десять хвилин: зонд міг
	// зникнути разом із ним, і без цього хост лишився б без бекапів
	// назавжди.
	reap := time.NewTicker(2 * time.Minute)
	defer reap.Stop()

	s.log.Info("диспетчер збору конфігів запущено", "інтервал", every)

	for {
		select {
		case <-ctx.Done():
			return

		case <-reap.C:
			if n, err := s.store.ReapStuckJobs(ctx, 10*time.Minute); err != nil {
				s.log.Warn("прибирання завислих завдань", "err", err)
			} else if n > 0 {
				s.log.Warn("завдання збору зависли й позначені як невдалі", "рядків", n)
			}
			s.reapCommandRuns(ctx)
			// Відкат прибирається в тому ж такті, але його «зависло»
			// означає інше, ніж у решти: зонд міг зникнути ПОСЕРЕД
			// заливки, і на пристрої лишився стан, якого не описує
			// жоден із двох конфігів. Тому там не відмова, а
			// контрольний збір — див. store.ReapStuckRollbacks.
			s.reapRollbacks(ctx)

		case <-t.C:
			// Закриття доведених до кінця прогонів — на швидкому такті
			// й ДО перевірки, чи є живі зонди. Прогін, у якому останній
			// хост щойно відзвітував (або жоден і не мав зонда),
			// інакше висів би «у роботі» до прибиральника, а сторінка
			// весь цей час перечитувала б його щодві секунди.
			s.settleCommandRuns(ctx)
			// Підсумок контрольних зборів після відкатів — теж на
			// швидкому такті й теж до перевірки живих зондів: намір,
			// у якого перевірка вже зібралась, інакше висів би
			// «перевіряємо» до наступного прибиральника, а сторінка
			// весь цей час його перечитувала б.
			s.settleRollbacks(ctx)

			online := s.onlineAgentIDs()
			if len(online) == 0 {
				continue
			}

			// Прохання «розпізнати зараз» їде тим самим тактом:
			// це той самий шлях REST → база → жива сесія, і власний
			// таймер задля кількох рядків був би зайвою деталлю.
			s.dispatchIdentify(ctx, online)

			// Масове виконання команд — так само: та сама черга в базі,
			// та сама жива сесія. Власний такт додав би ще один
			// інтервал, який довелося б тримати узгодженим із цим.
			s.dispatchCommandJobs(ctx, online)

			// Відкат — тим самим тактом і тією ж живою сесією. Власної
			// черги йому не потрібно: у .proto для нього є окреме
			// повідомлення, а шлях «база → сесія» той самий.
			s.dispatchRollbacks(ctx, online)

			jobs, err := s.store.ClaimConfigJobs(ctx, online, 16, s.ring)
			if err != nil {
				s.log.Error("вибірка завдань збору", "err", err)
				continue
			}

			for _, j := range jobs {
				ok := s.PushToAgent(j.AgentID, &npv1.ControlDown{
					Payload: &npv1.ControlDown_ConfigJob{ConfigJob: j.Job},
				})
				if !ok {
					// Сесія обірвалась між вибіркою й відправкою.
					// Повертаємо в чергу, а не втрачаємо: наступний тік
					// віддасть завдання, щойно зонд повернеться.
					_ = s.store.FinishConfigJob(ctx, j.JobID, "failed",
						"зонд відключився до надсилання завдання", "")
					continue
				}
				s.log.Info("завдання збору надіслано",
					"job_id", j.JobID, "agent", j.AgentID,
					"device", j.Job.GetDevice().GetName())
			}
		}
	}
}

// onlineAgentIDs — хто зараз на зв'язку.
func (s *Service) onlineAgentIDs() []string {
	s.mu.RLock()
	defer s.mu.RUnlock()

	out := make([]string, 0, len(s.sessions))
	for id := range s.sessions {
		out = append(out, id)
	}
	return out
}

// dispatchIdentify роздає прохання розпізнати хост.
//
// Зонд не виконує все, що прилетіло: DiscoveryRequest лише зрушує
// задачі, які вже є в його розкладі. Тому чек розпізнавання створюється
// раніше — у RequestIdentify, — а сюди доходить саме поштовх.
func (s *Service) dispatchIdentify(ctx context.Context, online []string) {
	reqs, err := s.store.ClaimIdentifyRequests(ctx, online, 64)
	if err != nil {
		s.log.Warn("вибірка прохань розпізнати", "err", err)
		return
	}
	if len(reqs) == 0 {
		return
	}

	// Спершу план, потім поштовх. RequestIdentify міг щойно завести
	// чек, а зонд зрушує лише те, що вже має в розкладі: без цього
	// рядка кнопка мовчки нічого не робила б саме там, де вона
	// найпотрібніша — на щойно заведеному хості.
	s.syncPlansTick(ctx)

	for _, r := range reqs {
		ok := s.PushToAgent(r.AgentID, &npv1.ControlDown{
			Payload: &npv1.ControlDown_DiscoveryRequest{
				DiscoveryRequest: &npv1.DiscoveryRequest{
					RunId:     "identify:" + r.DeviceID,
					DeviceIds: []string{r.DeviceID},
				},
			},
		})
		if !ok {
			s.log.Warn("зонд відключився до надсилання прохання розпізнати",
				"agent", r.AgentID, "device", r.DeviceID)
			continue
		}
		s.log.Info("розпізнавання зрушено вручну",
			"agent", r.AgentID, "device", r.DeviceID)
	}
}