Bridge-and-Join-s/internal/lkgateway/health.go

package lkgateway

import (
	"context"
	"fmt"
	"log"
	"net"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

// health.go — стабильность и самоконтроль сервиса.
//
// Три независимых слоя защиты, чтобы сбой не остался незамеченным:
//  1. superviseLoop — оборачивает каждую фоновую горутину: recover() ловит
//     панику, пишет в лог и в ленту, и перезапускает горутину. Паника одного
//     поллера больше не роняет весь процесс.
//  2. heartbeat — каждый поллер «отмечается» на каждом тике. Монитор следит за
//     свежестью отметок; если компонент завис (нет отметок > 3 интервалов) —
//     поднимает алерт в ленту «Новости» (оператор увидит).
//  3. sd_notify watchdog — пока все компоненты живы, процесс «пинает» systemd
//     (WATCHDOG=1). Если внутренний цикл завис намертво (дедлок) и отметки
//     прекратились — systemd по таймауту сам перезапустит весь сервис.

// compHealth — состояние одного наблюдаемого компонента.
type compHealth struct {
	Name     string
	Interval time.Duration // ожидаемый интервал между отметками
	LastBeat time.Time
	LastErr  string
	alerted  bool // алерт «завис» уже поднят — не дублируем
}

var (
	healthMu  sync.Mutex
	healthReg = map[string]*compHealth{}
)

// healthRegister объявляет компонент и его ожидаемый интервал отметок.
func healthRegister(name string, interval time.Duration) {
	healthMu.Lock()
	defer healthMu.Unlock()
	if _, ok := healthReg[name]; !ok {
		healthReg[name] = &compHealth{Name: name, Interval: interval, LastBeat: time.Now()}
	}
}

// healthBeat — компонент жив: обновляет отметку и снимает флаг алерта.
func healthBeat(name string) {
	healthMu.Lock()
	defer healthMu.Unlock()
	c := healthReg[name]
	if c == nil {
		c = &compHealth{Name: name, Interval: time.Minute}
		healthReg[name] = c
	}
	c.LastBeat = time.Now()
	c.alerted = false
}

// healthError фиксирует последнюю ошибку компонента (для UI; на watchdog не
// влияет — временная сетевая ошибка не должна перезапускать сервис).
func healthError(name, msg string) {
	healthMu.Lock()
	defer healthMu.Unlock()
	if c := healthReg[name]; c != nil {
		c.LastErr = msg
	}
}

// healthComponent возвращает зарегистрированный компонент по имени (или nil).
func healthComponent(name string) *compHealth {
	healthMu.Lock()
	defer healthMu.Unlock()
	return healthReg[name]
}

// HealthStatus — снимок состояния компонента для панели администратора.
type HealthStatus struct {
	Name     string
	Healthy  bool
	Age      time.Duration
	LastErr  string
	LastBeat time.Time
}

// HealthSnapshot возвращает состояние всех компонентов, отсортированное по имени.
// Healthy = свежая отметка (моложе 3 интервалов). LastErr на Healthy не влияет.
func HealthSnapshot() []HealthStatus {
	healthMu.Lock()
	defer healthMu.Unlock()
	out := make([]HealthStatus, 0, len(healthReg))
	now := time.Now()
	for _, c := range healthReg {
		age := now.Sub(c.LastBeat)
		out = append(out, HealthStatus{
			Name:     c.Name,
			Healthy:  age < 3*c.Interval,
			Age:      age,
			LastErr:  c.LastErr,
			LastBeat: c.LastBeat,
		})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
	return out
}

// superviseLoop запускает body в бесконечном цикле с recover(): паника или
// неожиданный выход body → лог + алерт + перезапуск через 5 с. Возвращается
// только когда ctx отменён (штатная остановка).
func superviseLoop(ctx context.Context, name string, rc *RuntimeConfig, body func(ctx context.Context)) {
	for {
		if ctx.Err() != nil {
			return
		}
		crashed := runGuarded(name, rc, func() { body(ctx) })
		if ctx.Err() != nil {
			return
		}
		if crashed {
			healthError(name, "перезапуск после сбоя")
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(5 * time.Second):
		}
	}
}

// runGuarded выполняет fn под recover(); true, если была паника.
func runGuarded(name string, rc *RuntimeConfig, fn func()) (crashed bool) {
	defer func() {
		if r := recover(); r != nil {
			crashed = true
			msg := fmt.Sprintf("%v", r)
			log.Printf("lk-gateway: КОМПОНЕНТ %s УПАЛ (паника): %s — перезапуск через 5с", name, msg)
			healthError(name, "паника: "+msg)
			raiseSystemAlert(rc, "Сбой компонента: "+name,
				fmt.Sprintf("Компонент «%s» аварийно завершился (%s) и автоматически перезапускается. Проверьте лог bj-server.", name, msg))
		}
	}()
	fn()
	return false
}

// superviseHealth — монитор здоровья + sd_notify watchdog. Раз в 15 с проверяет
// свежесть отметок всех компонентов: зависшие → алерт в ленту; если все живы —
// «пинает» systemd (при Type=notify + WatchdogSec). Запускается напрямую (сам
// себя не роняет — тело тривиально и без паник).
func (s *Server) superviseHealth(ctx context.Context) {
	sdNotify("READY=1")
	wd := watchdogTimeout()
	if wd > 0 {
		log.Printf("lk-gateway: systemd watchdog активен (таймаут %s)", wd)
	}
	ticker := time.NewTicker(15 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			allLive := true
			for _, h := range HealthSnapshot() {
				// Затяжное зависание (>6 интервалов): восстановить горутину не
				// удалось (recover бы уже перезапустил). Сообщаем и перезапускаем
				// весь процесс — systemd (Restart=always) поднимет за 5 с.
				if c := healthComponent(h.Name); c != nil && h.Age > 6*c.Interval {
					log.Printf("lk-gateway: КОМПОНЕНТ %s ЗАВИС НАМЕРТВО (%s без отметок) — перезапуск процесса", h.Name, h.Age.Round(time.Second))
					raiseSystemAlertSync(s.rc, "Аварийный перезапуск сервиса",
						fmt.Sprintf("Компонент «%s» завис (%s без признаков работы) и не восстановился. Сервис перезапускается автоматически.", h.Name, h.Age.Round(time.Second)))
					time.Sleep(300 * time.Millisecond)
					os.Exit(1)
				}
				if h.Healthy {
					continue
				}
				allLive = false
				s.alertStale(h)
			}
			// Пинаем watchdog только когда все компоненты живы (если включён
			// Type=notify + WatchdogSec — дополнительный внешний контроль).
			if allLive {
				sdNotify("WATCHDOG=1")
			}
		}
	}
}

// alertStale поднимает алерт «компонент завис» единожды (до восстановления).
func (s *Server) alertStale(h HealthStatus) {
	healthMu.Lock()
	c := healthReg[h.Name]
	if c == nil || c.alerted {
		healthMu.Unlock()
		return
	}
	c.alerted = true
	healthMu.Unlock()
	log.Printf("lk-gateway: КОМПОНЕНТ %s НЕ ОТВЕЧАЕТ (нет отметок %s)", h.Name, h.Age.Round(time.Second))
	raiseSystemAlert(s.rc, "Компонент не отвечает: "+h.Name,
		fmt.Sprintf("Фоновый компонент «%s» не подавал признаков работы %s. Сервис пытается восстановиться автоматически; если сообщение повторяется — проверьте bj-server.", h.Name, h.Age.Round(time.Second)))
}

// raiseSystemAlert добавляет системное событие в ленту «Новости» (дедуп по
// заголовку в пределах 10 минут, чтобы не засорять при флапе).
func raiseSystemAlert(rc *RuntimeConfig, title, body string) {
	if rc == nil {
		return
	}
	id := "sys-" + strconv.FormatInt(time.Now().Unix()/600, 10) + "-" + slugify(title)
	_ = rc.AddNews(NewsItem{
		ID:    id,
		At:    time.Now(),
		Kind:  "system",
		Title: title,
		Body:  body,
	})
	// Внешняя доставка (SMTP/Zabbix) — асинхронно, чтобы не блокировать монитор.
	go dispatchAlert(rc, title, body)
}

// raiseSystemAlertSync — как raiseSystemAlert, но с синхронной внешней доставкой:
// нужен на пути аварийного os.Exit, где горутины не успеют отработать.
func raiseSystemAlertSync(rc *RuntimeConfig, title, body string) {
	if rc == nil {
		return
	}
	id := "sys-" + strconv.FormatInt(time.Now().Unix()/600, 10) + "-" + slugify(title)
	_ = rc.AddNews(NewsItem{ID: id, At: time.Now(), Kind: "system", Title: title, Body: body})
	dispatchAlert(rc, title, body)
}

func slugify(s string) string {
	s = strings.ToLower(strings.TrimSpace(s))
	var b strings.Builder
	for _, r := range s {
		if r >= 'a' && r <= 'z' || r >= '0' && r <= '9' {
			b.WriteRune(r)
		} else if r == ' ' || r == '-' || r == ':' {
			b.WriteByte('-')
		}
	}
	out := strings.Trim(b.String(), "-")
	if out == "" {
		return "alert"
	}
	if len(out) > 40 {
		out = out[:40]
	}
	return out
}

// --- sd_notify (systemd) без внешних зависимостей ---

// sdNotify шлёт сообщение состояния в systemd через NOTIFY_SOCKET.
// No-op, если сокет не задан (запуск вне systemd).
func sdNotify(state string) {
	sock := os.Getenv("NOTIFY_SOCKET")
	if sock == "" {
		return
	}
	if strings.HasPrefix(sock, "@") { // абстрактный сокет
		sock = "\x00" + sock[1:]
	}
	conn, err := net.DialUnix("unixgram", nil, &net.UnixAddr{Name: sock, Net: "unixgram"})
	if err != nil {
		return
	}
	defer conn.Close()
	_, _ = conn.Write([]byte(state))
}

// watchdogTimeout возвращает интервал watchdog из WATCHDOG_USEC (0 — выключен).
func watchdogTimeout() time.Duration {
	v := os.Getenv("WATCHDOG_USEC")
	if v == "" {
		return 0
	}
	usec, err := strconv.ParseInt(v, 10, 64)
	if err != nil || usec <= 0 {
		return 0
	}
	return time.Duration(usec) * time.Microsecond
}