omnex/control-plane/api/internal/alerts/watcher.go

package alerts

import (
	"context"
	"fmt"
	"log"
	"time"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	metricsclient "k8s.io/metrics/pkg/client/clientset/versioned"
)

const pollInterval = 30 * time.Second

// Raisons de conteneur considérées comme des pannes (hors CrashLoopBackOff
// initial normal au tout premier démarrage d'un pod fraîchement créé — non
// distingué ici, le bruit au provisioning est acceptable).
var badWaitingReasons = map[string]bool{
	"CrashLoopBackOff":           true,
	"ImagePullBackOff":           true,
	"ErrImagePull":               true,
	"CreateContainerConfigError": true,
	"CreateContainerError":       true,
	"InvalidImageName":           true,
	"RunContainerError":          true,
}

type podKey struct{ namespace, pod string }

// Watcher scrute périodiquement les pods des namespaces actifs et notifie
// tout changement d'état (sain -> panne, panne -> rétabli), ainsi que les
// dépassements de seuil CPU/mémoire (voir resources.go) et de remplissage
// des volumes (PVC). Les destinataires (recipients) sont résolus à chaque
// tick où une notification est nécessaire — chaque admin configure son
// propre webhook Discord / bot Telegram dans son profil (voir
// internal/profile), rien n'est figé au démarrage.
type Watcher struct {
	k8sClient     *kubernetes.Clientset
	metricsClient *metricsclient.Clientset // optionnel : nil désactive les alertes CPU/mémoire (le reste du watcher continue de fonctionner)
	recipients    func(ctx context.Context) ([]Notifier, error)
	namespaces    func() ([]string, error)
	state         map[podKey]string // dernière raison de panne connue ("" = sain)
	resourceHigh  map[podKey]int    // ticks consécutifs au-dessus du seuil CPU/mémoire
	pvcHigh       map[pvcKey]int    // ticks consécutifs au-dessus du seuil de remplissage PVC
}

func NewWatcher(k8sClient *kubernetes.Clientset, metricsClient *metricsclient.Clientset, recipients func(ctx context.Context) ([]Notifier, error), namespaces func() ([]string, error)) *Watcher {
	return &Watcher{
		k8sClient:     k8sClient,
		metricsClient: metricsClient,
		recipients:    recipients,
		namespaces:    namespaces,
		state:         make(map[podKey]string),
		resourceHigh:  make(map[podKey]int),
		pvcHigh:       make(map[pvcKey]int),
	}
}

// Run boucle jusqu'à annulation du contexte. À lancer dans une goroutine.
func (w *Watcher) Run(ctx context.Context) {
	ticker := time.NewTicker(pollInterval)
	defer ticker.Stop()
	for {
		w.tick(ctx)
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func (w *Watcher) tick(ctx context.Context) {
	nsList, err := w.namespaces()
	if err != nil {
		log.Printf("alerts: liste des démos actives indisponible: %v", err)
		return
	}
	nsSet := make(map[string]bool, len(nsList))
	for _, ns := range nsList {
		nsSet[ns] = true
	}

	// Les destinataires ne sont chargés (requête DB) que si une notification
	// s'avère nécessaire ce tick-ci, et une seule fois par tick.
	var notifier Notifier
	notifierLoaded := false
	loadNotifier := func() Notifier {
		if notifierLoaded {
			return notifier
		}
		notifierLoaded = true
		list, err := w.recipients(ctx)
		if err != nil {
			log.Printf("alerts: chargement des destinataires échoué: %v", err)
			return nil
		}
		if len(list) == 0 {
			return nil
		}
		notifier = MultiNotifier(list)
		return notifier
	}

	seen := make(map[podKey]bool, len(w.state))
	seenHigh := make(map[podKey]bool, len(w.resourceHigh))
	nodeNames := make(map[string]bool) // nœuds hébergeant un pod Running d'un namespace actif (voir checkPVCUsage)

	for _, ns := range nsList {
		pods, err := w.k8sClient.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{})
		if err != nil {
			log.Printf("alerts: liste des pods de %s indisponible: %v", ns, err)
			continue
		}
		for _, pod := range pods.Items {
			key := podKey{namespace: ns, pod: pod.Name}
			seen[key] = true

			reason := problemReason(pod)
			prev, known := w.state[key]
			if reason != prev {
				if known {
					w.notify(ctx, loadNotifier(), key, prev, reason)
				} else if reason != "" {
					// Premier passage sur ce pod : on ne notifie que s'il est
					// déjà en panne (sinon rien d'anormal à signaler).
					w.notify(ctx, loadNotifier(), key, "", reason)
				}
				w.state[key] = reason
			}

			if pod.Status.Phase == corev1.PodRunning && pod.Spec.NodeName != "" {
				nodeNames[pod.Spec.NodeName] = true
			}
		}

		w.checkResourceUsage(ctx, loadNotifier, ns, pods.Items, seenHigh)
	}

	// Pods disparus (démo supprimée, rollout) : on oublie leur état pour ne
	// pas polluer la mémoire indéfiniment.
	for key := range w.state {
		if !seen[key] {
			delete(w.state, key)
		}
	}
	w.clearResourceRecovered(ctx, loadNotifier, seenHigh)
	w.checkPVCUsage(ctx, loadNotifier, nsSet, nodeNames)
}

func (w *Watcher) notify(ctx context.Context, notifier Notifier, key podKey, prev, reason string) {
	var msg string
	switch {
	case prev == "" && reason != "":
		msg = fmt.Sprintf("🔴 [%s] pod %s en erreur : %s", key.namespace, key.pod, reason)
	case prev != "" && reason == "":
		msg = fmt.Sprintf("🟢 [%s] pod %s rétabli", key.namespace, key.pod)
	default:
		msg = fmt.Sprintf("🔴 [%s] pod %s toujours en erreur : %s", key.namespace, key.pod, reason)
	}
	w.send(ctx, notifier, msg)
}

// problemReason retourne une raison de panne non vide si le pod est dans un
// état anormal (phase Failed, ou conteneur en CrashLoopBackOff/ImagePullBackOff/
// OOMKilled...), "" si tout va bien.
func problemReason(pod corev1.Pod) string {
	if pod.DeletionTimestamp != nil {
		return "" // suppression en cours, pas une panne
	}
	if pod.Status.Phase == corev1.PodFailed {
		if pod.Status.Reason != "" {
			return pod.Status.Reason
		}
		return "Failed"
	}
	for _, cs := range pod.Status.ContainerStatuses {
		if waiting := cs.State.Waiting; waiting != nil && badWaitingReasons[waiting.Reason] {
			return waiting.Reason
		}
		if term := cs.State.Terminated; term != nil && term.Reason == "OOMKilled" {
			return "OOMKilled"
		}
	}
	return ""
}