Netpulse_SasS/server/internal/grpcapi/ncm_commands.go

package grpcapi

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/json"
	"time"

	npv1 "github.com/netpulse/netpulse/gen/go/netpulse/v1"
	"github.com/netpulse/netpulse/server/internal/store"
)

// Масове виконання команд: серверна половина.
//
// Транспорту тут свого немає жодного — і в цьому вся суть. Завдання
// їде тим самим ConfigJob, вивід повертається тим самим стрімом
// UploadConfig, диспетчер працює в тому самому такті. Різниця в одному
// полі (config_type) і в тому, куди лягає результат.

// commandOutcome — те, що надсилає зонд (див. ncmx.CommandReport).
type commandOutcome struct {
	Command string `json:"command"`
	Output  string `json:"output"`
	Error   string `json:"error,omitempty"`
}

type commandReport struct {
	Outcomes []commandOutcome `json:"outcomes"`
	Error    string           `json:"error,omitempty"`
}

// isCommandUpload розрізняє вивід команд і зібраний конфіг.
//
// Заголовок від зонда — підказка, а не доказ: він приходить із поля, яке
// агент лише переспівує з завдання, і зонд старішої версії заповнить
// його як завгодно. Тому вирішує база: id завдання або є рядком прогону
// команд, або ні. Ціна — один запит за первинним ключем; ціна помилки —
// вивід `display cpu` у ролі версії конфігу в архіві.
func (s *Service) isCommandUpload(ctx context.Context, header *npv1.ConfigHeader) bool {
	if header.GetJobId() == "" {
		return false
	}
	if header.GetConfigType() != store.CommandConfigType {
		// Швидкий шлях для звичайного бекапу: зайвий запит на кожен
		// зібраний конфіг нічого не дав би.
		return false
	}
	ok, err := s.store.IsCommandTarget(ctx, header.GetJobId())
	if err != nil {
		s.log.Warn("звірка завдання з прогоном команд", "job", header.GetJobId(), "err", err)
		return false
	}
	return ok
}

// storeCommandResult кладе вивід команд у рядок прогону.
//
// Повертає готову квитанцію, а не помилку: зонд має отримати відповідь
// у будь-якому разі — інакше він вважатиме вивантаження незавершеним і
// повторить його разом із новою сесією до пристрою.
func (s *Service) storeCommandResult(ctx context.Context, header *npv1.ConfigHeader,
	body []byte, tr *npv1.ConfigTrailer) *npv1.ConfigReceipt {

	targetID := header.GetJobId()

	if !tr.GetSuccess() {
		// Зонд не дійшов до команд: не під'єднався, не дочекався
		// привітання. Показувати нічого, крім причини й стенограми, —
		// і саме вони тут найпотрібніші.
		msg := tr.GetError().GetMessage()
		if msg == "" {
			msg = "зонд не зміг виконати команди"
		}
		if err := s.store.FinishCommandTarget(ctx, targetID, "failed", msg, nil,
			tr.GetTranscript()); err != nil {
			s.log.Error("запис невдалого виконання команд", "target", targetID, "err", err)
		}
		return &npv1.ConfigReceipt{JobId: targetID, Accepted: true, Error: tr.GetError()}
	}

	plain, err := decodeBody(body, header.GetEncoding())
	if err != nil {
		_ = s.store.FinishCommandTarget(ctx, targetID, "failed",
			"не вдалося розпакувати вивід: "+err.Error(), nil, tr.GetTranscript())
		return &npv1.ConfigReceipt{
			JobId: targetID, Accepted: false,
			Error: &npv1.Error{Code: "bad_encoding", Message: err.Error()},
		}
	}

	// Сума звіряється так само, як для конфігу: обірваний посеред
	// передачі JSON розібрався б із помилкою, а обірваний на межі
	// елемента — мовчки, і хост лишився б із половиною виводу, схожою
	// на цілий.
	if want := tr.GetContentSha256(); len(want) > 0 {
		got := sha256.Sum256(plain)
		if !bytes.Equal(got[:], want) {
			_ = s.store.FinishCommandTarget(ctx, targetID, "failed",
				"тіло не відповідає заявленому sha256", nil, tr.GetTranscript())
			return &npv1.ConfigReceipt{
				JobId: targetID, Accepted: false,
				Error: &npv1.Error{
					Code: "checksum_mismatch", Retryable: true,
					Message: "тіло не відповідає заявленому sha256",
				},
			}
		}
	}

	var rep commandReport
	if err := json.Unmarshal(plain, &rep); err != nil {
		_ = s.store.FinishCommandTarget(ctx, targetID, "failed",
			"незрозумілий формат виводу від зонда: "+err.Error(), nil, tr.GetTranscript())
		return &npv1.ConfigReceipt{
			JobId: targetID, Accepted: false,
			Error: &npv1.Error{Code: "bad_body", Message: err.Error()},
		}
	}

	// Підготовчі команди профілю приїхали разом із командами людини —
	// зонд про різницю не знає й знати не мусить. Розмічаємо тут, за
	// переліком самого прогону.
	userCmds, err := s.store.CommandRunCommands(ctx, targetID)
	if err != nil {
		s.log.Warn("перелік команд прогону", "target", targetID, "err", err)
	}
	isUser := make(map[string]bool, len(userCmds))
	for _, c := range userCmds {
		isUser[c] = true
	}

	outcomes := make([]store.CommandOutcome, 0, len(rep.Outcomes))
	for _, o := range rep.Outcomes {
		outcomes = append(outcomes, store.CommandOutcome{
			Command: o.Command,
			Output:  o.Output,
			Error:   o.Error,
			Prep:    len(isUser) > 0 && !isUser[o.Command],
		})
	}

	status := "success"
	if rep.Error != "" {
		status = "failed"
	}
	if err := s.store.FinishCommandTarget(ctx, targetID, status, rep.Error,
		outcomes, tr.GetTranscript()); err != nil {
		s.log.Error("запис результату команд", "target", targetID, "err", err)
	}

	s.log.Info("вивід команд прийнято",
		"target", targetID, "команд", len(outcomes), "стан", status)

	return &npv1.ConfigReceipt{JobId: targetID, Accepted: true}
}

// ---------------------------------------------------------------------
// Диспетчер
// ---------------------------------------------------------------------

// dispatchCommandJobs роздає хости прогонів живим сесіям.
//
// Їде в такті збору конфігів, а не власним таймером: це той самий шлях
// «база → жива сесія», і другий цикл із власним інтервалом додав би ще
// одне місце, де налаштування розходяться.
func (s *Service) dispatchCommandJobs(ctx context.Context, online []string) {
	jobs, err := s.store.ClaimCommandJobs(ctx, online, 16, s.ring)
	if err != nil {
		s.log.Error("вибірка команд для виконання", "err", err)
		return
	}

	for _, j := range jobs {
		ok := s.PushToAgent(j.AgentID, &npv1.ControlDown{
			Payload: &npv1.ControlDown_ConfigJob{ConfigJob: j.Job},
		})
		if !ok {
			// Сесія обірвалась між вибіркою й відправкою. Хост падає, а
			// не повертається в чергу: на пристрої ще нічого не
			// відбувалось, але мовчазний повтор на зонді, який зараз
			// відвалюється, крутив би прогін нескінченно. Один рядок із
			// причиною — і людина сама вирішує, чи повторювати.
			_ = s.store.FinishCommandTarget(ctx, j.TargetID, "failed",
				"зонд відключився до надсилання завдання", nil, "")
			continue
		}
		if err := s.store.MarkCommandTargetSent(ctx, j.TargetID); err != nil {
			s.log.Warn("позначка про надсилання команд", "target", j.TargetID, "err", err)
		}
		s.log.Info("команди надіслано зонду",
			"target", j.TargetID, "agent", j.AgentID, "device", j.DeviceName)
	}
}

// reapCommandRuns прибирає зависле.
//
// Рідкісним тактом навмисно: обидва випадки — це хвилини очікування, і
// перевіряти їх щоп'ять секунд означало б платити запитом за подію, якої
// в нормальній роботі не буває.
func (s *Service) reapCommandRuns(ctx context.Context) {
	if n, err := s.store.ReapStuckCommandTargets(ctx, 2*time.Minute); err != nil {
		s.log.Warn("прибирання завислих команд", "err", err)
	} else if n > 0 {
		s.log.Warn("зависле виконання команд перезаведено або закрито", "рядків", n)
	}
}

// settleCommandRuns закриває прогони, у яких не лишилось роботи.
func (s *Service) settleCommandRuns(ctx context.Context) {
	if n, err := s.store.SettleCommandRuns(ctx); err != nil {
		s.log.Warn("закриття прогонів команд", "err", err)
	} else if n > 0 {
		s.log.Info("прогони команд завершено", "прогонів", n)
	}
}