chore: update
ci-api / test (push) Successful in 28m2s

This commit is contained in:
Xor290
2026-08-08 18:27:52 +02:00
parent 000ed75313
commit 2f6e5a953d
15 changed files with 939 additions and 42 deletions
@@ -0,0 +1,199 @@
package alerts
import (
"context"
"encoding/json"
"fmt"
"log"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
const (
// Seuils d'alerte, en % de la limite (CPU/mémoire) ou de la capacité (PVC).
resourceHighThresholdPct = 90
pvcFullThresholdPct = 85
// Nombre de ticks consécutifs au-dessus du seuil avant notification
// (pollInterval = 30s, soit ~90s) : absorbe les pics transitoires (déploiement
// en cours, backup Velero, migration...) sans notifier à tort.
consecutiveTicksBeforeAlert = 3
)
// send envoie un message si un notifier est disponible, journalise en cas d'échec.
func (w *Watcher) send(ctx context.Context, notifier Notifier, msg string) {
if notifier == nil {
return
}
if err := notifier.Notify(ctx, msg); err != nil {
log.Printf("alerts: notification échouée: %v", err)
}
}
// podResourcePercents calcule le % d'usage CPU/mémoire d'un pod par rapport
// aux limites déclarées dans son spec. ok=false si un conteneur n'a pas de
// limite définie (pourcentage non significatif dans ce cas — on ne peut pas
// dire à quel point il est "plein").
func podResourcePercents(pod corev1.Pod, cpuMilli, memBytes int64) (cpuPct, memPct int, ok bool) {
var cpuLimit, memLimit int64
for _, c := range pod.Spec.Containers {
cl := c.Resources.Limits.Cpu()
ml := c.Resources.Limits.Memory()
if cl.IsZero() || ml.IsZero() {
return 0, 0, false
}
cpuLimit += cl.MilliValue()
memLimit += ml.Value()
}
if cpuLimit == 0 || memLimit == 0 {
return 0, 0, false
}
return int(cpuMilli * 100 / cpuLimit), int(memBytes * 100 / memLimit), true
}
// checkResourceUsage alerte si un pod dépasse resourceHighThresholdPct de CPU
// ou de mémoire pendant consecutiveTicksBeforeAlert ticks d'affilée. seenHigh
// est alimenté pour la purge de récupération en fin de tick (voir
// clearResourceRecovered).
func (w *Watcher) checkResourceUsage(ctx context.Context, loadNotifier func() Notifier, ns string, pods []corev1.Pod, seenHigh map[podKey]bool) {
if w.metricsClient == nil {
return
}
list, err := w.metricsClient.MetricsV1beta1().PodMetricses(ns).List(ctx, metav1.ListOptions{})
if err != nil {
return // metrics-server temporairement indisponible : pas une panne à notifier
}
usage := make(map[string][2]int64, len(list.Items)) // pod -> [cpuMilli, memBytes]
for _, m := range list.Items {
var cpu, mem int64
for _, c := range m.Containers {
cpu += c.Usage.Cpu().MilliValue()
mem += c.Usage.Memory().Value()
}
usage[m.Name] = [2]int64{cpu, mem}
}
for _, pod := range pods {
u, ok := usage[pod.Name]
if !ok {
continue
}
cpuPct, memPct, ok := podResourcePercents(pod, u[0], u[1])
if !ok || (cpuPct < resourceHighThresholdPct && memPct < resourceHighThresholdPct) {
continue
}
key := podKey{namespace: ns, pod: pod.Name}
seenHigh[key] = true
w.resourceHigh[key]++
if w.resourceHigh[key] == consecutiveTicksBeforeAlert {
w.send(ctx, loadNotifier(), fmt.Sprintf(
"🟠 [%s] pod %s : usage élevé — CPU %d%%, mémoire %d%% (limite)", ns, pod.Name, cpuPct, memPct))
}
}
}
// clearResourceRecovered notifie le retour sous le seuil CPU/mémoire et purge
// l'état des pods qui ne sont plus au-dessus du seuil ce tick-ci.
func (w *Watcher) clearResourceRecovered(ctx context.Context, loadNotifier func() Notifier, seenHigh map[podKey]bool) {
for key, count := range w.resourceHigh {
if seenHigh[key] {
continue
}
if count >= consecutiveTicksBeforeAlert {
w.send(ctx, loadNotifier(), fmt.Sprintf(
"🟢 [%s] pod %s : usage CPU/mémoire revenu à la normale", key.namespace, key.pod))
}
delete(w.resourceHigh, key)
}
}
type pvcKey struct{ namespace, pvc string }
// kubeletSummary : sous-ensemble de l'API kubelet /stats/summary (voir
// k8s.io/kubelet/pkg/apis/stats/v1alpha1.Summary sur
// github.com/kubernetes/kubelet) utilisé pour lire l'usage réel des volumes
// (PVC) montés sur un nœud — aucune métrique de ce type n'est exposée par
// metrics-server, qui ne couvre que CPU/mémoire. Nécessite la permission
// "get" sur le sous-type "nodes/proxy" pour le compte utilisé par l'API
// (aucun ClusterRole scoping n'existe dans ce repo à ce jour — vérifier si
// un jour le control-plane tourne avec un ServiceAccount restreint).
type kubeletSummary struct {
Pods []struct {
PodRef struct {
Namespace string `json:"namespace"`
} `json:"podRef"`
Volume []struct {
CapacityBytes *uint64 `json:"capacityBytes,omitempty"`
UsedBytes *uint64 `json:"usedBytes,omitempty"`
PVCRef *struct {
Name string `json:"name"`
Namespace string `json:"namespace"`
} `json:"pvcRef,omitempty"`
} `json:"volume,omitempty"`
} `json:"pods"`
}
// fetchNodeSummary récupère les stats kubelet d'un nœud via le proxy API
// server (équivalent de `kubectl get --raw /api/v1/nodes/<node>/proxy/stats/summary`).
func (w *Watcher) fetchNodeSummary(ctx context.Context, nodeName string) (*kubeletSummary, error) {
raw, err := w.k8sClient.CoreV1().RESTClient().Get().
Resource("nodes").
Name(nodeName).
SubResource("proxy").
Suffix("stats/summary").
Do(ctx).Raw()
if err != nil {
return nil, err
}
var summary kubeletSummary
if err := json.Unmarshal(raw, &summary); err != nil {
return nil, err
}
return &summary, nil
}
// checkPVCUsage alerte si un volume (PVC) d'un namespace actif dépasse
// pvcFullThresholdPct de remplissage pendant consecutiveTicksBeforeAlert
// ticks. nodeNames : nœuds hébergeant au moins un pod Running d'un namespace
// actif (voir tick) — on n'interroge que ceux-là, pas tout le cluster.
func (w *Watcher) checkPVCUsage(ctx context.Context, loadNotifier func() Notifier, nsSet map[string]bool, nodeNames map[string]bool) {
seenHigh := make(map[pvcKey]bool)
for node := range nodeNames {
summary, err := w.fetchNodeSummary(ctx, node)
if err != nil {
log.Printf("alerts: stats kubelet indisponibles pour le nœud %s: %v", node, err)
continue
}
for _, pod := range summary.Pods {
if !nsSet[pod.PodRef.Namespace] {
continue
}
for _, vol := range pod.Volume {
if vol.PVCRef == nil || vol.CapacityBytes == nil || vol.UsedBytes == nil || *vol.CapacityBytes == 0 {
continue
}
pct := int(*vol.UsedBytes * 100 / *vol.CapacityBytes)
if pct < pvcFullThresholdPct {
continue
}
key := pvcKey{namespace: vol.PVCRef.Namespace, pvc: vol.PVCRef.Name}
seenHigh[key] = true
w.pvcHigh[key]++
if w.pvcHigh[key] == consecutiveTicksBeforeAlert {
w.send(ctx, loadNotifier(), fmt.Sprintf(
"🟠 [%s] volume %s rempli à %d%%", key.namespace, key.pvc, pct))
}
}
}
}
for key, count := range w.pvcHigh {
if seenHigh[key] {
continue
}
if count >= consecutiveTicksBeforeAlert {
w.send(ctx, loadNotifier(), fmt.Sprintf(
"🟢 [%s] volume %s repassé sous %d%% de remplissage", key.namespace, key.pvc, pvcFullThresholdPct))
}
delete(w.pvcHigh, key)
}
}
@@ -0,0 +1,90 @@
package alerts
import (
"testing"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
)
func podWithLimits(cpuLimit, memLimit string) corev1.Pod {
return corev1.Pod{
Spec: corev1.PodSpec{
Containers: []corev1.Container{
{
Resources: corev1.ResourceRequirements{
Limits: corev1.ResourceList{
corev1.ResourceCPU: resource.MustParse(cpuLimit),
corev1.ResourceMemory: resource.MustParse(memLimit),
},
},
},
},
},
}
}
func TestPodResourcePercents(t *testing.T) {
pod := podWithLimits("500m", "256Mi")
cpuPct, memPct, ok := podResourcePercents(pod, 450, 256*1024*1024)
if !ok {
t.Fatal("ok = false, want true")
}
if cpuPct != 90 {
t.Errorf("cpuPct = %d, want 90", cpuPct)
}
if memPct != 100 {
t.Errorf("memPct = %d, want 100", memPct)
}
}
func TestPodResourcePercents_NoLimitsDefined(t *testing.T) {
pod := corev1.Pod{
Spec: corev1.PodSpec{
Containers: []corev1.Container{{}},
},
}
_, _, ok := podResourcePercents(pod, 100, 100)
if ok {
t.Error("ok = true, want false when container has no resource limits")
}
}
func TestPodResourcePercents_MultiContainerSumsLimits(t *testing.T) {
pod := corev1.Pod{
Spec: corev1.PodSpec{
Containers: []corev1.Container{
{
Resources: corev1.ResourceRequirements{
Limits: corev1.ResourceList{
corev1.ResourceCPU: resource.MustParse("250m"),
corev1.ResourceMemory: resource.MustParse("128Mi"),
},
},
},
{
Resources: corev1.ResourceRequirements{
Limits: corev1.ResourceList{
corev1.ResourceCPU: resource.MustParse("250m"),
corev1.ResourceMemory: resource.MustParse("128Mi"),
},
},
},
},
},
}
// Limite totale : 500m CPU, 256Mi mémoire.
cpuPct, memPct, ok := podResourcePercents(pod, 250, 128*1024*1024)
if !ok {
t.Fatal("ok = false, want true")
}
if cpuPct != 50 {
t.Errorf("cpuPct = %d, want 50", cpuPct)
}
if memPct != 50 {
t.Errorf("memPct = %d, want 50", memPct)
}
}
+38 -19
View File
@@ -9,6 +9,7 @@ import (
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
@@ -29,23 +30,31 @@ var badWaitingReasons = map[string]bool{
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). 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.
// 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
recipients func(ctx context.Context) ([]Notifier, error)
namespaces func() ([]string, error)
state map[podKey]string // dernière raison de panne connue ("" = sain)
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, recipients func(ctx context.Context) ([]Notifier, error), namespaces func() ([]string, error)) *Watcher {
func NewWatcher(k8sClient *kubernetes.Clientset, metricsClient *metricsclient.Clientset, recipients func(ctx context.Context) ([]Notifier, error), namespaces func() ([]string, error)) *Watcher {
return &Watcher{
k8sClient: k8sClient,
recipients: recipients,
namespaces: namespaces,
state: make(map[podKey]string),
k8sClient: k8sClient,
metricsClient: metricsClient,
recipients: recipients,
namespaces: namespaces,
state: make(map[podKey]string),
resourceHigh: make(map[podKey]int),
pvcHigh: make(map[pvcKey]int),
}
}
@@ -69,6 +78,10 @@ func (w *Watcher) tick(ctx context.Context) {
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.
@@ -92,6 +105,9 @@ func (w *Watcher) tick(ctx context.Context) {
}
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 {
@@ -114,7 +130,13 @@ func (w *Watcher) tick(ctx context.Context) {
}
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
@@ -124,12 +146,11 @@ func (w *Watcher) tick(ctx context.Context) {
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) {
if notifier == nil {
return
}
var msg string
switch {
case prev == "" && reason != "":
@@ -139,9 +160,7 @@ func (w *Watcher) notify(ctx context.Context, notifier Notifier, key podKey, pre
default:
msg = fmt.Sprintf("🔴 [%s] pod %s toujours en erreur : %s", key.namespace, key.pod, reason)
}
if err := notifier.Notify(ctx, msg); err != nil {
log.Printf("alerts: notification échouée: %v", err)
}
w.send(ctx, notifier, msg)
}
// problemReason retourne une raison de panne non vide si le pod est dans un