Netpulse_SasS/agent/internal/session/apply_jobs.go

package session

import (
	"context"
	"time"

	"github.com/netpulse/netpulse/agent/internal/ncmx"
	npv1 "github.com/netpulse/netpulse/gen/go/netpulse/v1"
	"google.golang.org/protobuf/types/known/durationpb"
)

// runApplyJob заливає конфіг на пристрій і звітує про результат.
//
// Виконується в окремій горутині — з тієї ж причини, що й збір конфігу:
// заливка триває хвилини, і тримати на ній контрольний цикл означало б
// не відповідати на ping і бути визнаним мертвим саме тоді, коли зонд
// робить найнебезпечнішу роботу.
//
// Жодної власної логіки погодження тут немає й не буде. Завдання, що
// приїхало в межах живої автентифікованої сесії, виконується — усе
// рішення про те, чи можна його виконувати, ухвалено на сервері
// (ncm.rollbacks: awaiting_approval → approved). Зонд, який намагався б
// вирішувати це вдруге, мав би власну копію політики — і вона розійшлася
// б із серверною першого ж дня.
func (s *Session) runApplyJob(ctx context.Context, out chan<- *npv1.ControlUp,
	job *npv1.ConfigApplyJob) {

	log := s.log.With("rollback", job.GetRollbackId(),
		"device", job.GetDevice().GetName(),
		"рядків", len(job.GetCommands()))

	// Warn, а не Info, і до початку роботи. Запис на живе залізо — те,
	// що шукають у журналі зонда після аварії, і рядок «почали» має
	// стояти навіть тоді, коли після нього не буде жодного іншого.
	log.Warn("заливка конфігу почалась")

	res := ncmx.Apply(ctx, job)

	result := &npv1.ConfigApplyResult{
		RollbackId: job.GetRollbackId(),
		Success:    res.Err == nil,
		Outcomes:   res.Outcomes,
		Committed:  res.Committed,
		Transcript: res.Transcript,
		Duration:   durationpb.New(res.Duration),
	}
	if res.Err != nil {
		result.Error = &npv1.Error{Code: "apply_failed", Message: res.Err.Error()}
		log.Error("заливка конфігу не завершилась",
			"помилка", res.Err, "виконано_рядків", len(res.Outcomes))
	} else {
		log.Warn("заливка конфігу завершена",
			"рядків", len(res.Outcomes), "збережено", res.Committed,
			"тривалість", res.Duration.Round(time.Millisecond))
	}

	s.sendApplyResult(ctx, out, result)
}

// sendApplyResult віддає звіт, чекаючи місця в черзі.
//
// Тут свідомо НЕ enqueue. Той кидає повідомлення, коли черга забита, і
// для heartbeat це правильно: сервер помітить пропажу за таймаутом.
// Звіт про заливку загубити не можна за жодних умов — сервер, який його
// не отримав, знає лише, що на пристрій щось пішло, і не знає, чим
// скінчилось.
//
// Якщо ж канал усе-таки помер разом із сесією, звіт справді зникне — і
// саме на цей випадок на сервері стоїть прибиральник, який переводить
// таке завдання не у відмову, а в контрольний збір: піти й подивитись,
// що реально стало на пристрої. Здогадуватись тут нема про що, пристрій
// поруч.
func (s *Session) sendApplyResult(ctx context.Context, out chan<- *npv1.ControlUp,
	result *npv1.ConfigApplyResult) {

	msg := &npv1.ControlUp{
		Seq:     s.seq.Add(1),
		Payload: &npv1.ControlUp_ConfigApplyResult{ConfigApplyResult: result},
	}

	select {
	case out <- msg:
	case <-ctx.Done():
		s.log.Error("звіт про заливку не поїхав: сесія обірвалась",
			"rollback", result.GetRollbackId())
	case <-time.After(30 * time.Second):
		s.log.Error("звіт про заливку не поїхав: контрольний канал забитий",
			"rollback", result.GetRollbackId())
	}
}