Netpulse_SasS/agent/internal/session/config_jobs.go

package session

import (
	"bytes"
	"compress/gzip"
	"context"
	"errors"
	"time"

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

// errNoConnection — сесія обірвалась, поки збирали конфіг.
var errNoConnection = errors.New("немає живого зʼєднання з сервером")

// chunkSize — розмір шматка вивантаження.
//
// 64 КБ: помітно менше за типовий ліміт gRPC-повідомлення, тому навіть
// конфіг на кілька МБ проходить без налаштування транспорту, і при
// цьому не так дрібно, щоб платити накладними за кожен рядок.
const chunkSize = 64 << 10

// runConfigJob знімає конфіг і вивантажує його на сервер.
//
// Виконується в окремій горутині: збір конфігу з великого шасі триває
// хвилини, і тримати на ньому контрольний цикл означало б не відповідати
// на ping і бути визнаним мертвим саме тоді, коли агент найбільше
// зайнятий корисною роботою.
func (s *Session) runConfigJob(ctx context.Context, job *npv1.ConfigJob) {
	log := s.log.With("job_id", job.GetJobId(),
		"device", job.GetDevice().GetName(),
		"config_type", job.GetConfigType())

	// Масове виконання команд їде тим самим завданням і тим самим
	// стрімом — інакше довелося б заводити другий шлях сервер→зонд
	// заради того, що вже працює. Різниця лише в тому, що робити з
	// виводом, і про неї каже config_type.
	//
	// Гілка тут, а не в ncmx: транспорт, вхід і пошук запрошення в обох
	// випадках однакові, і розводити їх нижче означало б розводити й це.
	if job.GetConfigType() == commandConfigType {
		s.runCommandJob(ctx, job)
		return
	}

	// Конфіг-файли самої машини — теж це саме завдання й те саме
	// вивантаження, лише замість сесії до пристрою читається власний
	// диск. Розвилка за наявністю local_files, а не за config_type:
	// config_type тут означає назву зрізу в архіві, і навантажувати
	// його ще й способом збору означало б, що перейменування зрізу
	// тихо змінює те, звідки беруться дані.
	if job.GetLocalFiles() != nil {
		s.runLocalFilesJob(ctx, job)
		return
	}

	log.Info("збір конфігу почався")

	res, err := ncmx.Collect(ctx, job)
	if err != nil {
		log.Error("збір конфігу", "помилка", err)
		s.uploadFailure(ctx, job, err, res.Transcript)
		return
	}

	log.Info("конфіг знято",
		"байтів", len(res.Body), "рядків", res.LineCount,
		"тривалість", res.Duration.Round(time.Millisecond))

	if err := s.uploadConfig(ctx, job, res); err != nil {
		log.Error("вивантаження конфігу", "помилка", err)
	}
}

func (s *Session) uploadConfig(ctx context.Context, job *npv1.ConfigJob, res ncmx.Result) error {
	return s.uploadConfigAs(ctx, job, res, "")
}

// uploadConfigAs — те саме вивантаження, але з відбитком машини.
//
// Відбиток заповнює лише збір локальних файлів: для комутатора питання
// «з якої машини це знято» не має сенсу — знято з пристрою, а зонд лише
// набирав. Для сервера воно головне, бо зонд і є той сервер.
func (s *Session) uploadConfigAs(ctx context.Context, job *npv1.ConfigJob,
	res ncmx.Result, machineID string) error {

	conn := s.client.Load()
	if conn == nil {
		return errNoConnection
	}

	stream, err := (*conn).UploadConfig(ctx)
	if err != nil {
		return err
	}

	// Стискаємо завжди: конфіги — це текст із величезною надлишковістю,
	// а канал до зонда часто вузький. Сервер розуміє обидва варіанти,
	// тому вибір тут наш.
	var zbuf bytes.Buffer
	zw := gzip.NewWriter(&zbuf)
	if _, err := zw.Write(res.Body); err != nil {
		return err
	}
	if err := zw.Close(); err != nil {
		return err
	}
	payload := zbuf.Bytes()

	if err := stream.Send(&npv1.ConfigUpload{
		Part: &npv1.ConfigUpload_Header{Header: &npv1.ConfigHeader{
			JobId:       job.GetJobId(),
			AgentId:     s.cfg.AgentID,
			DeviceId:    job.GetDevice().GetDeviceId(),
			ConfigType:  job.GetConfigType(),
			CollectedAt: timestamppb.Now(),
			Encoding:    "gzip",
			MachineId:   machineID,
		}},
	}); err != nil {
		return err
	}

	var chunks uint32
	for off := 0; off < len(payload); off += chunkSize {
		end := off + chunkSize
		if end > len(payload) {
			end = len(payload)
		}
		if err := stream.Send(&npv1.ConfigUpload{
			Part: &npv1.ConfigUpload_Chunk{Chunk: &npv1.ConfigChunk{
				Sequence: chunks,
				Data:     payload[off:end],
			}},
		}); err != nil {
			return err
		}
		chunks++
	}

	// sha256 рахується від ОРИГІНАЛУ, до стиснення: сервер звіряє саме
	// тіло конфігу, а не його пакування. Інакше зміна рівня стиснення
	// виглядала б як зміна конфігу.
	if err := stream.Send(&npv1.ConfigUpload{
		Part: &npv1.ConfigUpload_Trailer{Trailer: &npv1.ConfigTrailer{
			Success:       true,
			ContentSha256: res.SHA256,
			SizeBytes:     uint64(len(res.Body)),
			LineCount:     uint32(res.LineCount),
			ChunkCount:    chunks,
			Duration:      durationpb.New(res.Duration),
			Transcript:    res.Transcript,
		}},
	}); err != nil {
		return err
	}

	receipt, err := stream.CloseAndRecv()
	if err != nil {
		return err
	}
	if !receipt.GetAccepted() {
		s.log.Warn("сервер не прийняв конфіг",
			"job_id", job.GetJobId(), "причина", receipt.GetError().GetMessage())
		return nil
	}

	s.log.Info("конфіг прийнято",
		"job_id", job.GetJobId(),
		"змінився", !receipt.GetUnchanged(),
		"commit", receipt.GetCommitSha())
	return nil
}

// uploadFailure повідомляє про невдачу тим самим стрімом.
//
// Мовчати не можна: без цього невдалий збір виглядає на сервері як
// «завдання ще виконується», і наступний запуск за розкладом лише
// додасть другий такий самий. Транскрипт долучається саме тут — він
// потрібен рівно тоді, коли щось пішло не так.
func (s *Session) uploadFailure(ctx context.Context, job *npv1.ConfigJob, cause error, transcript string) {
	conn := s.client.Load()
	if conn == nil {
		return
	}
	stream, err := (*conn).UploadConfig(ctx)
	if err != nil {
		return
	}

	_ = stream.Send(&npv1.ConfigUpload{
		Part: &npv1.ConfigUpload_Header{Header: &npv1.ConfigHeader{
			JobId:       job.GetJobId(),
			AgentId:     s.cfg.AgentID,
			DeviceId:    job.GetDevice().GetDeviceId(),
			ConfigType:  job.GetConfigType(),
			CollectedAt: timestamppb.Now(),
			Encoding:    "none",
		}},
	})
	_ = stream.Send(&npv1.ConfigUpload{
		Part: &npv1.ConfigUpload_Trailer{Trailer: &npv1.ConfigTrailer{
			Success: false,
			Error: &npv1.Error{
				Code:    "collect_failed",
				Message: cause.Error(),
			},
			Transcript: transcript,
		}},
	})
	_, _ = stream.CloseAndRecv()
}