feat(rules,alerting): 평가 매니저와 발송 연결 — 알림이 실제로 나간다

internal/alerting/alerting.go

@@ -0,0 +1,29 @@
+// Package alerting 은 룰 평가와 발송을 이어 붙인다.
+//
+// storage 가 tsdb 와 promql 을 잇는 것과 같은 자리다 — rules 는 전송 방식을
+// 모르고 notify 는 룰을 모른다. 둘 다 상대를 모르면 누군가는 알아야 하고,
+// 그게 여기다.
+package alerting
+
+import (
+	"github.com/KeiaiLab/nodevitals-observatory/internal/notify"
+	"github.com/KeiaiLab/nodevitals-observatory/internal/rules"
+)
+
+// NewSink 는 Notifier 를 rules.Sink 로 감싼다.
+func NewSink(n *notify.Notifier) rules.Sink { return &sink{n: n} }
+
+type sink struct{ n *notify.Notifier }
+
+func (s *sink) Notify(ns []rules.Notification) error {
+	alerts := make([]notify.Alert, 0, len(ns))
+	for _, n := range ns {
+		alerts = append(alerts, notify.Alert{
+			Labels:      n.Labels,
+			Annotations: n.Annotations,
+			Status:      notify.Status(n.Status),
+			Value:       n.Value,
+		})
+	}
+	return s.n.Send(alerts)
+}

internal/alerting/alerting_test.go

@@ -0,0 +1,123 @@
+package alerting
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/KeiaiLab/nodevitals-observatory/internal/labels"
+	"github.com/KeiaiLab/nodevitals-observatory/internal/notify"
+	"github.com/KeiaiLab/nodevitals-observatory/internal/rules"
+	"github.com/KeiaiLab/nodevitals-observatory/internal/storage"
+	"github.com/KeiaiLab/nodevitals-observatory/internal/tsdb"
+)
+
+// 룰 파일 → 로드 → 평가 → 발송까지 한 줄로 흐르는지 본다. 조각을 따로
+// 검증한 뒤 남는 위험은 이음매뿐이고, 이 경로가 곧 제품의 알림 기능이다.
+func TestRuleFileToWebhook(t *testing.T) {
+	var (
+		mu     sync.Mutex
+		bodies [][]byte
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		buf := make([]byte, r.ContentLength)
+		_, _ = r.Body.Read(buf)
+		mu.Lock()
+		bodies = append(bodies, buf)
+		mu.Unlock()
+		w.WriteHeader(200)
+	}))
+	t.Cleanup(srv.Close)
+
+	// 1) 룰 파일
+	dir := t.TempDir()
+	rf := filepath.Join(dir, "hw.yaml")
+	if err := os.WriteFile(rf, []byte(`groups:
+  - name: hardware
+    interval: 30s
+    rules:
+      - alert: GpuTooHot
+        expr: gpu_temperature_celsius > 85
+        for: 0s
+        labels:
+          severity: critical
+        annotations:
+          summary: "GPU 과열"
+`), 0o600); err != nil {
+		t.Fatalf("WriteFile: %v", err)
+	}
+
+	groups, err := rules.LoadDir(dir)
+	if err != nil {
+		t.Fatalf("LoadDir: %v", err)
+	}
+
+	// 2) 데이터
+	db, err := tsdb.Open(tsdb.DefaultOptions(t.TempDir()))
+	if err != nil {
+		t.Fatalf("tsdb.Open: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	lset := labels.LabelsFromMap(map[string]string{
+		labels.MetricName: "gpu_temperature_celsius",
+		"node":            "e106",
+		"gpu":             "0",
+	})
+	if err := db.Append(lset, 1000, 91); err != nil {
+		t.Fatalf("Append: %v", err)
+	}
+
+	// 3) 조립
+	n := notify.New(notify.Config{URL: srv.URL, Timeout: 2 * time.Second})
+	m := rules.NewManager(groups, storage.New(db), NewSink(n))
+
+	// 4) 평가
+	m.EvalOnce(1000)
+
+	mu.Lock()
+	defer mu.Unlock()
+	if len(bodies) != 1 {
+		t.Fatalf("발송 %d 건, want 1", len(bodies))
+	}
+	var payload struct {
+		Alerts []struct {
+			Labels      map[string]string `json:"labels"`
+			Annotations map[string]string `json:"annotations"`
+			Status      string            `json:"status"`
+			Value       float64           `json:"value"`
+		} `json:"alerts"`
+	}
+	if err := json.Unmarshal(bodies[0], &payload); err != nil {
+		t.Fatalf("페이로드: %v (%q)", err, bodies[0])
+	}
+	if len(payload.Alerts) != 1 {
+		t.Fatalf("알림 %d 건: %+v", len(payload.Alerts), payload)
+	}
+	a := payload.Alerts[0]
+	if a.Labels["alertname"] != "GpuTooHot" {
+		t.Errorf("alertname=%q", a.Labels["alertname"])
+	}
+	if a.Labels["severity"] != "critical" {
+		t.Errorf("severity=%q", a.Labels["severity"])
+	}
+	if a.Labels["node"] != "e106" || a.Labels["gpu"] != "0" {
+		t.Errorf("원본 라벨이 유지되지 않았다: %+v", a.Labels)
+	}
+	if a.Annotations["summary"] != "GPU 과열" {
+		t.Errorf("summary=%q", a.Annotations["summary"])
+	}
+	if a.Status != "firing" || a.Value != 91 {
+		t.Errorf("status=%q value=%v", a.Status, a.Value)
+	}
+
+	// 5) 상태가 안 바뀌면 다시 보내지 않는다
+	m.EvalOnce(2000)
+	if len(bodies) != 1 {
+		t.Errorf("재평가 후 발송 %d 건, want 1", len(bodies))
+	}
+}

internal/rules/manager.go

@@ -0,0 +1,190 @@
+package rules
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/KeiaiLab/nodevitals-observatory/internal/promql"
+)
+
+// Notification 은 밖으로 나갈 알림 하나다. 발송 구현(webhook·그 밖)을 이
+// 패키지가 알지 못하도록 자체 타입을 쓴다 — 룰 평가가 전송 방식에 묶이면
+// 채널을 바꿀 때마다 평가기를 건드리게 된다.
+type Notification struct {
+	Labels      map[string]string
+	Annotations map[string]string
+	Status      string // "firing" | "resolved"
+	Value       float64
+}
+
+// Sink 는 알림을 받아 어딘가로 보낸다.
+type Sink interface {
+	Notify([]Notification) error
+}
+
+// RuleState·GroupState 는 /api/v1/rules 응답의 원천이다.
+type RuleState struct {
+	Name       string
+	Expression string
+	State      string
+	LastError  string
+	Alerts     []Notification
+}
+
+type GroupState struct {
+	Name  string
+	Rules []RuleState
+}
+
+// Manager 는 그룹들을 주기적으로 평가하고, 상태가 바뀐 알림만 내보낸다.
+type Manager struct {
+	mu     sync.Mutex
+	groups []*Group
+	q      promql.Queryable
+	sink   Sink
+	log    *slog.Logger
+
+	// firing 은 직전 평가에서 발송된 알림들이다. 이것과 비교해 "새로 뜬 것"
+	// 과 "풀린 것" 만 내보낸다 — 매 평가마다 같은 알림을 넘기면 반복 억제
+	// 판단이 발송기 쪽으로만 몰려 흐려진다.
+	firing map[string]Notification
+}
+
+func NewManager(groups []*Group, q promql.Queryable, sink Sink) *Manager {
+	return &Manager{
+		groups: groups, q: q, sink: sink,
+		log:    slog.Default(),
+		firing: map[string]Notification{},
+	}
+}
+
+// SetQueryable 은 데이터 원천을 바꾼다(테스트·재구성용).
+func (m *Manager) SetQueryable(q promql.Queryable) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.q = q
+}
+
+// SetLogger 는 로거를 바꾼다.
+func (m *Manager) SetLogger(l *slog.Logger) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.log = l
+}
+
+// EvalOnce 는 모든 그룹을 한 번 평가하고 변화분을 내보낸다.
+func (m *Manager) EvalOnce(evalMS int64) {
+	m.mu.Lock()
+	q, sink, log := m.q, m.sink, m.log
+	groups := m.groups
+	m.mu.Unlock()
+
+	current := map[string]Notification{}
+	for _, g := range groups {
+		for _, r := range g.Rules {
+			for _, a := range r.Eval(q, evalMS) {
+				if a.State != StateFiring {
+					continue // pending 은 아직 사람에게 알릴 단계가 아니다
+				}
+				n := Notification{
+					Labels: a.Labels, Annotations: a.Annotations,
+					Status: "firing", Value: a.Value,
+				}
+				current[fingerprint(a.Labels)] = n
+			}
+		}
+	}
+
+	m.mu.Lock()
+	var out []Notification
+	for fp, n := range current {
+		if _, already := m.firing[fp]; !already {
+			out = append(out, n)
+		}
+	}
+	for fp, prev := range m.firing {
+		if _, still := current[fp]; !still {
+			r := prev
+			r.Status = "resolved"
+			out = append(out, r)
+		}
+	}
+	m.firing = current
+	m.mu.Unlock()
+
+	if len(out) == 0 {
+		return
+	}
+	// 발송 실패는 평가를 멈추지 않는다 — 수신 측 장애가 관측까지 세우면
+	// 정작 복구에 필요한 데이터가 끊긴다.
+	if err := sink.Notify(out); err != nil {
+		log.Warn("알림 발송 실패 — 평가는 계속한다", "건수", len(out), "err", err)
+	}
+}
+
+// Run 은 가장 짧은 그룹 주기로 평가를 반복한다. ctx 가 끝나면 돌아온다.
+func (m *Manager) Run(ctx context.Context) {
+	interval := m.tick()
+	t := time.NewTicker(interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case now := <-t.C:
+			m.EvalOnce(now.UnixMilli())
+		}
+	}
+}
+
+// tick 은 그룹 주기 중 가장 짧은 것이다. 그룹별 주기를 따로 돌리는 것은
+// 룰 수가 늘고 주기가 실제로 갈릴 때 하면 된다.
+func (m *Manager) tick() time.Duration {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	shortest := time.Duration(0)
+	for _, g := range m.groups {
+		if g.Interval <= 0 {
+			continue
+		}
+		if shortest == 0 || g.Interval < shortest {
+			shortest = g.Interval
+		}
+	}
+	if shortest == 0 {
+		shortest = 30 * time.Second
+	}
+	return shortest
+}
+
+// Snapshot 은 현재 룰 상태다. /api/v1/rules 가 이것을 그대로 낸다.
+func (m *Manager) Snapshot() []GroupState {
+	m.mu.Lock()
+	groups := m.groups
+	m.mu.Unlock()
+
+	out := make([]GroupState, 0, len(groups))
+	for _, g := range groups {
+		gs := GroupState{Name: g.Name}
+		for _, r := range g.Rules {
+			rs := RuleState{Name: r.Name, Expression: r.Expression(), State: "inactive"}
+			for _, a := range r.Active() {
+				rs.Alerts = append(rs.Alerts, Notification{
+					Labels: a.Labels, Annotations: a.Annotations,
+					Status: a.State.String(), Value: a.Value,
+				})
+				// firing 이 하나라도 있으면 룰 상태는 firing 이다.
+				if a.State == StateFiring {
+					rs.State = "firing"
+				} else if rs.State != "firing" {
+					rs.State = "pending"
+				}
+			}
+			gs.Rules = append(gs.Rules, rs)
+		}
+		out = append(out, gs)
+	}
+	return out
+}