From 49cd9f0190f598454de8ef3f59d3ccc9feba9eb3 Mon Sep 17 00:00:00 2001
From: phil <phil@keiailab.com>
Date: Wed, 12 Aug 2026 23:07:19 +0900
Subject: [PATCH] =?UTF-8?q?feat(rules,alerting):=20=ED=8F=89=EA=B0=80=20?=
 =?UTF-8?q?=EB=A7=A4=EB=8B=88=EC=A0=80=EC=99=80=20=EB=B0=9C=EC=86=A1=20?=
 =?UTF-8?q?=EC=97=B0=EA=B2=B0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

매니저가 그룹을 주기로 평가하고 **변화분만** 내보낸다.

- pending 은 보내지 않는다 — 아직 사람에게 알릴 단계가 아니다.
- 직전 firing 집합과 비교해 새로 뜬 것과 풀린 것만 낸다. 매 평가마다 같은
  알림을 넘기면 반복 억제 판단이 발송기 쪽으로만 몰려 흐려진다.
- 풀린 알림은 resolved 로 한 번 보낸다.
- **발송 실패는 평가를 멈추지 않는다.** 수신 측 장애가 관측까지 세우면
  정작 복구에 필요한 데이터가 끊긴다.

internal/alerting 은 rules 와 notify 를 잇는다 — storage 가 tsdb 와 promql 을
잇는 것과 같은 자리다. rules 는 전송 방식을 모르고 notify 는 룰을 모른다.

end-to-end 테스트: 룰 파일 → 로드 → 평가 → webhook. 조각을 따로 검증한 뒤
남는 위험은 이음매뿐이고, 이 경로가 곧 제품의 알림 기능이다.
---
 internal/alerting/alerting.go      |  29 +++++
 internal/alerting/alerting_test.go | 123 +++++++++++++++++++
 internal/rules/manager.go          | 190 +++++++++++++++++++++++++++++
 internal/rules/manager_test.go     | 145 ++++++++++++++++++++++
 4 files changed, 487 insertions(+)
 create mode 100644 internal/alerting/alerting.go
 create mode 100644 internal/alerting/alerting_test.go
 create mode 100644 internal/rules/manager.go
 create mode 100644 internal/rules/manager_test.go

diff --git a/internal/alerting/alerting.go b/internal/alerting/alerting.go
new file mode 100644
index 0000000..9413702
--- /dev/null
+++ b/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)
+}
diff --git a/internal/alerting/alerting_test.go b/internal/alerting/alerting_test.go
new file mode 100644
index 0000000..625bce6
--- /dev/null
+++ b/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))
+	}
+}
diff --git a/internal/rules/manager.go b/internal/rules/manager.go
new file mode 100644
index 0000000..08c11b4
--- /dev/null
+++ b/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
+}
diff --git a/internal/rules/manager_test.go b/internal/rules/manager_test.go
new file mode 100644
index 0000000..09595db
--- /dev/null
+++ b/internal/rules/manager_test.go
@@ -0,0 +1,145 @@
+package rules
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/KeiaiLab/nodevitals-observatory/internal/promql"
+)
+
+type recorder struct {
+	mu    sync.Mutex
+	sends [][]Notification
+}
+
+func (r *recorder) Notify(ns []Notification) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	cp := make([]Notification, len(ns))
+	copy(cp, ns)
+	r.sends = append(r.sends, cp)
+	return nil
+}
+
+func (r *recorder) all() []Notification {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var out []Notification
+	for _, s := range r.sends {
+		out = append(out, s...)
+	}
+	return out
+}
+
+func mgrWith(t *testing.T, q promql.Queryable, rec *recorder, rules ...*Rule) *Manager {
+	t.Helper()
+	g := &Group{Name: "g", Interval: time.Second, Rules: rules}
+	return NewManager([]*Group{g}, q, rec)
+}
+
+// firing 이 된 알림만 나간다 — pending 은 아직 사람에게 알릴 단계가 아니다.
+func TestManagerSendsOnlyFiring(t *testing.T) {
+	q := newQ(t, sample{"x", nil, 0, 5})
+	rec := &recorder{}
+	r := mustRule(t, "A", `x > 3`, time.Minute) // for 1분 → 첫 평가는 pending
+	m := mgrWith(t, q, rec, r)
+
+	m.EvalOnce(0)
+	if got := rec.all(); len(got) != 0 {
+		t.Fatalf("pending 인데 발송됐다: %+v", got)
+	}
+
+	m.EvalOnce(60_000) // for 충족
+	got := rec.all()
+	if len(got) != 1 || got[0].Status != "firing" {
+		t.Fatalf("발송 %+v, want firing 하나", got)
+	}
+}
+
+// 해소되면 resolved 를 한 번 보낸다 — 보내지 않으면 사람이 계속 대응한다.
+func TestManagerSendsResolved(t *testing.T) {
+	up := newQ(t, sample{"x", nil, 0, 5})
+	down := newQ(t, sample{"x", nil, 0, 1})
+	rec := &recorder{}
+	r := mustRule(t, "A", `x > 3`, 0)
+
+	g := &Group{Name: "g", Interval: time.Second, Rules: []*Rule{r}}
+	m := NewManager([]*Group{g}, up, rec)
+	m.EvalOnce(0)
+
+	m.SetQueryable(down)
+	m.EvalOnce(1000)
+
+	got := rec.all()
+	if len(got) != 2 {
+		t.Fatalf("발송 %d 건, want 2 (firing + resolved): %+v", len(got), got)
+	}
+	if got[1].Status != "resolved" {
+		t.Errorf("두 번째 발송 status=%q, want resolved", got[1].Status)
+	}
+}
+
+// 같은 알림이 계속 firing 이어도 반복해서 보내지 않는다 — 반복 억제는
+// 발송기가 하지만, 매니저가 매 평가마다 같은 것을 넘기면 그 판단이 흐려진다.
+func TestManagerDoesNotResendUnchangedFiring(t *testing.T) {
+	q := newQ(t, sample{"x", nil, 0, 5})
+	rec := &recorder{}
+	r := mustRule(t, "A", `x > 3`, 0)
+	m := mgrWith(t, q, rec, r)
+
+	m.EvalOnce(0)
+	m.EvalOnce(1000)
+	m.EvalOnce(2000)
+
+	if got := rec.all(); len(got) != 1 {
+		t.Errorf("발송 %d 건, want 1 (상태가 안 바뀌면 다시 보내지 않는다): %+v", len(got), got)
+	}
+}
+
+// 발송이 실패해도 평가는 계속돼야 한다 — 수신 측 장애가 관측을 멈추면 안 된다.
+type failingSink struct{ calls int }
+
+func (f *failingSink) Notify([]Notification) error { f.calls++; return errSink }
+
+var errSink = &sinkError{}
+
+type sinkError struct{}
+
+func (e *sinkError) Error() string { return "보낼 수 없다" }
+
+func TestManagerKeepsEvaluatingWhenSinkFails(t *testing.T) {
+	q := newQ(t, sample{"x", nil, 0, 5})
+	f := &failingSink{}
+	r := mustRule(t, "A", `x > 3`, 0)
+	g := &Group{Name: "g", Interval: time.Second, Rules: []*Rule{r}}
+	m := NewManager([]*Group{g}, q, f)
+
+	m.EvalOnce(0)
+	m.EvalOnce(1000)
+
+	if len(r.Active()) != 1 {
+		t.Errorf("발송 실패 후 활성 알림 %d, want 1 (평가는 계속된다)", len(r.Active()))
+	}
+}
+
+// API 노출용 스냅샷.
+func TestManagerSnapshotExposesRuleState(t *testing.T) {
+	q := newQ(t, sample{"x", nil, 0, 5})
+	rec := &recorder{}
+	r := mustRule(t, "A", `x > 3`, 0)
+	m := mgrWith(t, q, rec, r)
+	m.EvalOnce(0)
+
+	snap := m.Snapshot()
+	if len(snap) != 1 || len(snap[0].Rules) != 1 {
+		t.Fatalf("스냅샷 %+v", snap)
+	}
+	rs := snap[0].Rules[0]
+	if rs.Name != "A" || rs.State != "firing" {
+		t.Errorf("룰 상태 %+v", rs)
+	}
+	if rs.Expression == "" {
+		t.Error("식이 비어 있다")
+	}
+}