148 lines
3.2 KiB
Go
148 lines
3.2 KiB
Go
// Package alerts evaluates operational conditions without including credentials in notifications.
|
|
package alerts
|
|
|
|
import (
|
|
"context"
|
|
"linkport/internal/gateway"
|
|
"linkport/internal/store"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
type Source interface {
|
|
HealthSnapshot() (int, []gateway.HealthResult)
|
|
Metrics(string) gateway.Metrics
|
|
}
|
|
type Engine struct {
|
|
store Repository
|
|
rules conditionEvaluator
|
|
mu sync.Mutex
|
|
lifecycle sync.Mutex
|
|
cancel context.CancelFunc
|
|
done chan struct{}
|
|
notifier alertNotifier
|
|
}
|
|
|
|
func New(s Repository, source Source) *Engine {
|
|
return &Engine{store: s, rules: conditionEvaluator{source: source, samples: map[string][]sample{}}, notifier: newWebhookNotifier()}
|
|
}
|
|
func (e *Engine) Start() {
|
|
e.lifecycle.Lock()
|
|
defer e.lifecycle.Unlock()
|
|
if e.cancel != nil {
|
|
return
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
e.cancel = cancel
|
|
e.done = make(chan struct{})
|
|
go func() {
|
|
defer close(e.done)
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
e.Check(ctx)
|
|
e.Deliver(ctx)
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
func (e *Engine) Close() {
|
|
e.lifecycle.Lock()
|
|
defer e.lifecycle.Unlock()
|
|
if e.cancel != nil {
|
|
e.cancel()
|
|
<-e.done
|
|
e.cancel = nil
|
|
}
|
|
}
|
|
|
|
// Check only updates in-app conditions. It never contacts configured webhooks.
|
|
func (e *Engine) Check(ctx context.Context) error {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
return e.check(ctx, time.Now().UTC())
|
|
}
|
|
func (e *Engine) check(ctx context.Context, now time.Time) error {
|
|
c, err := e.store.AlertSettings()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ps, err := e.store.List()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ts, err := e.store.Targets()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
seen := map[string]bool{}
|
|
put := func(a store.Alert, active bool) error {
|
|
seen[a.Key] = true
|
|
return e.store.ReconcileAlert(a, active, c.WebhookEnabled, now)
|
|
}
|
|
if err = e.rules.evaluate(ctx, c, ps, ts, now, put); err != nil {
|
|
return err
|
|
}
|
|
assets, err := e.store.Assets()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = evaluateAssetCertificates(assets, c, now, put); err != nil {
|
|
return err
|
|
}
|
|
old, err := e.store.Alerts()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, a := range old {
|
|
if !seen[a.Key] && a.Status == "active" {
|
|
if err = e.store.ReconcileAlert(a, false, c.WebhookEnabled, now); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return e.store.PruneAlerts(now)
|
|
}
|
|
|
|
// Deliver is called only by the worker after the operator explicitly enables notifications.
|
|
// Each transition gets at most three attempts, separated by worker ticks.
|
|
func (e *Engine) Deliver(ctx context.Context) error {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
c, err := e.store.AlertSettings()
|
|
if err != nil || !c.WebhookEnabled {
|
|
return err
|
|
}
|
|
items, err := e.store.Alerts()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, a := range items {
|
|
if a.DeliveryStatus != "pending" {
|
|
continue
|
|
}
|
|
if err = ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
sent, err := e.notifier.Send(ctx, c, a)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
status := "pending"
|
|
if sent {
|
|
status = "sent"
|
|
}
|
|
if status != "sent" && a.DeliveryAttempts >= 2 {
|
|
status = "failed"
|
|
}
|
|
if err = e.store.AlertDelivery(a.ID, a.UpdatedAt, status); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|