Skip to content
Open
2 changes: 2 additions & 0 deletions cmd/ongrid/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -2940,6 +2940,7 @@ func main() {
PromQuerier: alertPromQuerier,
LogQuerier: lokiLogClient,
LogSearcher: logsBackendSvc,
ConsoleURL: cfg.PublicURL,
DeviceIdentityResolver: func(ctx context.Context, deviceID uint64) (managerbizalert.DeviceIdentity, error) {
device, err := deviceUC.Get(ctx, deviceID)
if err != nil {
Expand All @@ -2964,6 +2965,7 @@ func main() {
Usecase: alertUC,
MaxAttempts: 5,
Tick: cfg.Alert.EvaluatorInterval,
ConsoleURL: cfg.PublicURL,
Log: log.With(slog.String("comp", "alert-retry")),
})
eg.Go(func() error { return retryWorker.Loop(egCtx) })
Expand Down
19 changes: 17 additions & 2 deletions internal/manager/biz/alert/pipeline.go
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,10 @@ type PipelineEvaluatorOpts struct {

DeviceIdentityResolver DeviceIdentityResolver

// ConsoleURL is the public console base used for notification card deep links
// (devices / incidents). Empty disables action buttons.
ConsoleURL string

Log *slog.Logger
Now func() time.Time
}
Expand All @@ -117,6 +121,7 @@ type PipelineEvaluator struct {
logq LogQuerier
logSearcher logquery.Searcher
deviceIdentity DeviceIdentityResolver
consoleURL string

// gaugeSnapshot is the previous tick's (device_id, device_name) set
// used by refreshDeviceStalenessGauge to garbage-collect series for
Expand Down Expand Up @@ -166,6 +171,7 @@ func NewPipelineEvaluator(opts PipelineEvaluatorOpts) *PipelineEvaluator {
logq: opts.LogQuerier,
logSearcher: opts.LogSearcher,
deviceIdentity: opts.DeviceIdentityResolver,
consoleURL: strings.TrimRight(strings.TrimSpace(opts.ConsoleURL), "/"),
log: opts.Log,
now: opts.Now,
}
Expand Down Expand Up @@ -423,16 +429,24 @@ func (e *PipelineEvaluator) notify(ctx context.Context, res *FiringResult, summa
OccurredAt: at,
Labels: map[string]string{
"rule": res.Incident.Rule,
"rule_name": res.Incident.RuleName,
"status": res.Incident.Status,
"incident_id": fmt.Sprintf("%d", res.Incident.ID),
},
}
if res.Incident.RunbookURL != "" {
msg.Labels["runbook_url"] = res.Incident.RunbookURL
}
if e.consoleURL != "" {
msg.Labels["console_url"] = e.consoleURL
}
// Preserve application identity for notification routing and correlation.
// Control labels such as incident_id/rule still come from the incident.
if labels, err := res.Incident.Labels(); err != nil {
e.log.Warn("alert: decode notification identity failed", slog.Uint64("incident_id", res.Incident.ID), slog.Any("err", err))
} else {
for _, key := range []string{"service", "service_namespace", "deployment_environment_name", "service_instance_id", "span_name"} {
if value, ok := labels[key]; ok {
for _, key := range []string{"service", "service_namespace", "deployment_environment_name", "service_instance_id", "span_name", "rule_name", "rule_expr"} {
if value, ok := labels[key]; ok && msg.Labels[key] == "" {
msg.Labels[key] = value
}
}
Expand All @@ -446,6 +460,7 @@ func (e *PipelineEvaluator) notify(ctx context.Context, res *FiringResult, summa
e.log.Warn("alert: resolve device identity for notification failed",
slog.Uint64("device_id", deviceID), slog.Any("err", err))
} else if display := deviceDisplay(identity); display != "" {
msg.Labels["device_name"] = display
msg.Subject = strings.ReplaceAll(msg.Subject,
fmt.Sprintf("device_id=%d", deviceID), "device="+display)
if identity.Hostname != "" {
Expand Down
24 changes: 24 additions & 0 deletions internal/manager/biz/alert/pipeline_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,30 @@ func TestPipelineNotificationPreservesApplicationIdentity(t *testing.T) {
}
}

func TestPipelineNotificationInjectsConsoleURL(t *testing.T) {
repo, notifier := newFakeRepo(), &fakeNotifier{}
prom := &fakePromQuerier{result: &promquery.InstantResult{ResultType: "vector", Result: json.RawMessage(`[{"metric":{"device_id":"9"},"value":[1,"1"]}]`)}}
rules := NewStaticRulesProvider(WithMetricRawRules([]MetricRawRule{{ID: 1, RuleKey: "cpu_high", Name: "CPU high", ScopeType: "global", Expr: "up > 0", Severity: "critical"}}))
evaluator := newPipelineEvaluator(t, repo, notifier, rules, PipelineEvaluatorOpts{
PromQuerier: prom,
ConsoleURL: "https://ongrid.example.com/",
})
evaluator.EvaluateOnce(t.Context())
if len(notifier.msgs) != 1 {
t.Fatalf("notifications=%d", len(notifier.msgs))
}
labels := notifier.msgs[0].Labels
if labels["console_url"] != "https://ongrid.example.com" {
t.Fatalf("console_url = %q, want trimmed public URL", labels["console_url"])
}
if labels["status"] != model.IncidentStatusOpen {
t.Fatalf("status = %q, want %q", labels["status"], model.IncidentStatusOpen)
}
if labels["rule_name"] != "CPU high" {
t.Fatalf("rule_name = %q", labels["rule_name"])
}
}

func (f *fakeEdgeLister) List(_ context.Context, _ edgebiz.ListFilter) ([]*edgemodel.Edge, error) {
return f.edges, f.err
}
Expand Down
24 changes: 20 additions & 4 deletions internal/manager/biz/alert/retry.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log/slog"
"strings"
"time"

model "github.com/ongridio/ongrid/internal/manager/model/alert"
Expand All @@ -23,8 +24,11 @@ type RetryWorkerOpts struct {
MaxAttempts uint32
BackoffPerAttempt time.Duration
Tick time.Duration
Log *slog.Logger
Now func() time.Time
// ConsoleURL is the public console base injected into retry payloads so
// Feishu cards retain the same deep-link buttons as first-send notifies.
ConsoleURL string
Log *slog.Logger
Now func() time.Time
}

// RetryWorker drains failed notification_deliveries, re-runs the underlying
Expand All @@ -38,6 +42,7 @@ type RetryWorker struct {
maxAttempts uint32
backoffPerAttempt time.Duration
tick time.Duration
consoleURL string
log *slog.Logger
now func() time.Time
}
Expand Down Expand Up @@ -67,6 +72,7 @@ func NewRetryWorker(opts RetryWorkerOpts) *RetryWorker {
maxAttempts: opts.MaxAttempts,
backoffPerAttempt: opts.BackoffPerAttempt,
tick: opts.Tick,
consoleURL: strings.TrimRight(strings.TrimSpace(opts.ConsoleURL), "/"),
log: opts.Log,
now: opts.Now,
}
Expand Down Expand Up @@ -158,7 +164,7 @@ func (w *RetryWorker) retryOne(ctx context.Context, d *model.Delivery, now time.
return
}

msg := buildIncidentMessage(incident, now)
msg := buildIncidentMessage(incident, now, w.consoleURL)
sentAt := now
finished := now
sendErr := w.notifier.Send(ctx, msg, channel.Name)
Expand Down Expand Up @@ -200,15 +206,25 @@ func (w *RetryWorker) retryOne(ctx context.Context, d *model.Delivery, now time.
}
}

func buildIncidentMessage(incident *model.Incident, now time.Time) notify.Message {
func buildIncidentMessage(incident *model.Incident, now time.Time, consoleURL string) notify.Message {
severity := notify.Severity(incident.Severity)
if severity == "" {
severity = notify.SeverityWarning
}
labels := map[string]string{
"rule": incident.Rule,
"status": incident.Status,
"incident_id": fmt.Sprintf("%d", incident.ID),
}
if incident.RuleName != "" {
labels["rule_name"] = incident.RuleName
}
if incident.RunbookURL != "" {
labels["runbook_url"] = incident.RunbookURL
}
if consoleURL != "" {
labels["console_url"] = consoleURL
}
if incident.DeviceID != nil {
labels["device_id"] = fmt.Sprintf("%d", *incident.DeviceID)
}
Expand Down
Loading
Loading