Netpulse_SasS/agent/internal/session/logs.go

package session

import (
	"context"
	"net"
	"strings"
	"time"

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

// maxLogBatch — стеля подій в одній пачці.
//
// Журнал не телеметрія: під час аварії він приходить сплеском, і
// відправляти по одній події означає накрутити тисячі викликів рівно
// тоді, коли мережа й так не в порядку.
const maxLogBatch = 500

// logFlush — як довго накопичувати пачку.
//
// Півсекунди — компроміс: тригер бекапу за «%SYS-5-CONFIG_I» має
// спрацьовувати відчутно швидше за хвилинний цикл, а слати кожен рядок
// окремо немає сенсу.
const logFlush = 500 * time.Millisecond

// logsLoop віддає накопичені події серверу.
//
// Окремий стрім, а не контрольний канал: сплеск логів під час аварії не
// має заважати heartbeat і командам. Саме тому в контракті StreamLogs
// існує окремо від Control.
// Syslog і трапи їдуть ОДНИМ стрімом, а не двома.
//
// Спокуса завести другий є: приймачі різні, порти різні, розбір різний.
// Але за межами зонда це та сама подія з мережі — вона лягає в сусідні
// таблиці, звіряється з тими самими подієвими правилами й приїжджає в
// той самий момент аварії. Другий стрім означав би другий комплект
// підтверджень, лімітів і черг переповнення — і два різні місця, у
// яких по-різному вирішено, що робити з пачкою, яку не вдалося
// відправити.
func (s *Session) logsLoop(ctx context.Context, client npv1.AgentServiceClient) error {
	sys, trp := s.cfg.Syslog, s.cfg.Traps
	if sys == nil && trp == nil {
		return nil
	}

	stream, err := client.StreamLogs(ctx)
	if err != nil {
		return err
	}

	// Читач підтверджень окремою горутиною: Send і Recv на одному
	// стрімі — єдине, що gRPC дозволяє робити паралельно.
	ackErr := make(chan error, 1)
	go func() {
		for {
			ack, err := stream.Recv()
			if err != nil {
				ackErr <- err
				return
			}
			if e := ack.GetError(); e != nil {
				s.log.Warn("сервер не прийняв журнал",
					"code", e.GetCode(), "msg", e.GetMessage())
				continue
			}
			// Ліміти задає сервер: він бачить картину по всіх зондах і
			// краще знає, що вважати шумом. Стеля на джерело спільна
			// для обох приймачів — шумить не протокол, а пристрій.
			if sys != nil {
				sys.ApplyAck(ack.GetMinSeverity(), ack.GetRateLimitPerSource())
			}
			if trp != nil {
				trp.ApplyAck(ack.GetRateLimitPerSource())
			}
		}
	}()

	// Один із приймачів може бути вимкнений, тому канали готовності
	// беремо через nil-заглушку: читання з nil-каналу блокується
	// назавжди, і саме це в select потрібно — гілка, яка ніколи не
	// спрацює, замість гілки, якої немає.
	var sysReady, trapReady <-chan struct{}
	if sys != nil {
		sysReady = sys.Ready()
	}
	if trp != nil {
		trapReady = trp.Ready()
	}

	ticker := time.NewTicker(logFlush)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return nil
		case err := <-ackErr:
			return err
		case <-sysReady:
			// Подія з'явилась — але не летимо одразу: даємо тіку
			// зібрати сусідів у ту саму пачку.
		case <-trapReady:
		case <-ticker.C:
		}

		var (
			entries []*npv1.SyslogEntry
			traps   []*npv1.SnmpTrap
			dropped uint64
		)
		if sys != nil {
			var d uint64
			entries, d = sys.Drain(maxLogBatch)
			dropped += d
		}
		if trp != nil {
			// Половина пачки на трапи — не арифметика, а пріоритет:
			// трапів за секунду на порядок менше, ніж рядків журналу,
			// і стеля тут потрібна лише на випадок шторму. Витіснити
			// журнал вони не мають.
			var d uint64
			traps, d = trp.Drain(maxLogBatch / 2)
			dropped += d
		}
		if len(entries) == 0 && len(traps) == 0 && dropped == 0 {
			continue
		}

		batch := &npv1.LogBatch{
			BatchId: s.nextBatch.Add(1),
			AgentId: s.cfg.AgentID,
			Syslog:  entries,
			Traps:   traps,
			Dropped: dropped,
		}
		if err := stream.Send(batch); err != nil {
			// Невідправлене повертаємо в чергу: наступна сесія
			// доставить. Порядок зберігається — журнал читають
			// хронологічно.
			if sys != nil {
				sys.Requeue(entries)
			}
			if trp != nil {
				trp.Requeue(traps)
			}
			return err
		}
	}
}

// resolveDeviceByIP шукає хост за адресою джерела.
//
// Зіставлення на зонді, а не на сервері: у сервера немає контексту
// мережі клієнта, а один і той самий приватний діапазон трапляється в
// десятках кабінетів. Зонд же має свіжий список своїх хостів.
//
// Не знайшли — не біда: подія все одно доїде з порожнім device_id і
// заповненим source_ip. Викидати журнал через незнайому адресу означало
// б утратити рівно те, що показує появу нового заліза в мережі.
func (s *Session) resolveDeviceByIP(ip string) string {
	if ip == "" {
		return ""
	}
	addr := net.ParseIP(ip)

	s.devMu.RLock()
	defer s.devMu.RUnlock()

	for id, d := range s.devices {
		target := strings.TrimSpace(d.GetAddress())
		if target == "" {
			continue
		}
		if target == ip {
			return id
		}
		// Адреса хоста може бути записана з портом або як FQDN —
		// порівнюємо ще й розібраний варіант.
		if addr != nil {
			if host, _, err := net.SplitHostPort(target); err == nil {
				if net.ParseIP(host).Equal(addr) {
					return id
				}
			}
			if net.ParseIP(target).Equal(addr) {
				return id
			}
		}
	}
	return ""
}