amadan
feat/alerts-sns
Squashed from feat/alerts-sns into main.
13 files changed,
+1005
−35
README.md+17 −0cli.go+39 −2docs/site/cli.md+9 −3go.mod+1 −0go.sum+2 −0internal/host/agent.go+4 −0internal/host/alertsink.go+336 −0internal/host/alertsink_test.go+451 −0internal/host/metrics_loop.go+11 −21internal/host/metrics_loop_test.go+13 −8internal/status/snapshot.go+36 −1internal/status/snapshot_test.go+41 −0statuslive_test.go+45 −0
CI passed — run details
diff --git a/README.md b/README.md| index 22a5c3dca..bee763f04 100644 |
| --- a/README.md |
| +++ b/README.md |
| @@ -689,6 +689,23 @@ for the full design. |
| - `CARLOS_ALERT_WEBHOOK` — optional; the agent POSTs the alert JSON here on |
| fire and on clear. Unset simply skips the webhook channel (keymail |
| delivery and the console UI channel are unaffected). |
| +- `CARLOS_ALERT_SNS_TOPIC_ARN` — optional; the agent publishes each fire and |
| + clear to this SNS topic as a plain-text email-sized message, signed with |
| + the box's instance role (which needs `sns:Publish` on the topic). The |
| + region is read from the ARN, not from `AWS_REGION`. Set it to the box's |
| + own region's alerts topic and anyone subscribed to that topic gets the |
| + email. Either sink works alone and both may be set. A failed send is |
| + logged and counted, never fatal. The SDK tries a retryable failure up to |
| + three times within ten seconds. After that, a later tick does not send |
| + that alert again, unless the same tick's bucket write also failed, which |
| + repeats the whole transition on the next tick. A value that is not an SNS |
| + topic ARN (a subscription ARN, for example) is logged and the sink stays |
| + off. So does a client that cannot be built, which `carlos status --live` |
| + shows as `alerts=sns-unavailable`. |
| +- With neither sink set, the box says so: the agent logs `NO SINK |
| + CONFIGURED` at start, and the box's private status snapshot carries |
| + `metrics.alerts` with both sinks false, which `carlos status --live` prints |
| + as `alerts=not-delivered`. |
| - `CARLOS_ALERT_DISK_PCT` / `CARLOS_ALERT_MEM_PCT` / `CARLOS_ALERT_CPU_PCT` — |
| optional threshold overrides for the built-in disk / memory / host-CPU |
| rules, as percent integers 0-100 (e.g. `95`). Anything else is ignored |
diff --git a/cli.go b/cli.go| index 9b96c907f..f96163eed 100644 |
| --- a/cli.go |
| +++ b/cli.go |
| @@ -3429,13 +3429,50 @@ func printLiveBoxes(w io.Writer, boxes []status.Box, now time.Time) { |
| fmt.Fprintf(w, "%-24s %-12s ok (no metrics block — pre-upgrade host)\n", b.Label, orDash(b.Region)) |
| continue |
| } |
| - fmt.Fprintf(w, "%-24s %-12s release=%s tunnel=%s mem=%s/%s load=%s disk-free=%s\n", |
| + fmt.Fprintf(w, "%-24s %-12s release=%s tunnel=%s mem=%s/%s load=%s disk-free=%s alerts=%s\n", |
| b.Label, orDash(b.Region), orDash(m.Release), orDash(m.Tunnel), |
| humanBytes(m.MemAvailBytes), humanBytes(m.MemTotalBytes), |
| - humanFloat(m.Load1), humanBytes(m.DiskFreeBytes)) |
| + humanFloat(m.Load1), humanBytes(m.DiskFreeBytes), alertsWord(m.Alerts)) |
| } |
| } |
| +// alertsWord renders a box's alert-sink answer for the live listing. "not |
| +// delivered" is spelled out because it is the state discussion #192 found |
| +// silent for weeks: a box whose own disk, memory and cpu rules have nowhere |
| +// to send. A box that predates the field did not say, so it is a dash. |
| +func alertsWord(a *status.AlertDelivery) string { |
| + if a == nil { |
| + return "-" |
| + } |
| + var sinks []string |
| + if a.Webhook { |
| + sinks = append(sinks, "webhook") |
| + } |
| + if a.SNS { |
| + sinks = append(sinks, "sns") |
| + } |
| + if len(sinks) == 0 && !a.SNSUnavailable { |
| + return "not-delivered" |
| + } |
| + out := strings.Join(sinks, "+") |
| + if a.SNSUnavailable { |
| + // A topic is configured but its client could not be built. Alone it |
| + // is the whole answer; next to a working sink it is a warning. |
| + if out == "" { |
| + out = "sns-unavailable" |
| + } else { |
| + out += " (sns-unavailable)" |
| + } |
| + } |
| + if a.WebhookFailures > 0 { |
| + out += fmt.Sprintf(" (webhook failed %d)", a.WebhookFailures) |
| + } |
| + if a.SNSFailures > 0 { |
| + out += fmt.Sprintf(" (sns failed %d)", a.SNSFailures) |
| + } |
| + return out |
| +} |
| + |
| // orDash, humanBytes and humanFloat render one field of a metrics block. |
| // Unknown is a dash in all three — never a zero, never a blank column: the |
| // status package carries "I couldn't measure this" as a nil pointer all the |
diff --git a/docs/site/cli.md b/docs/site/cli.md| index 115129a74..c08054481 100644 |
| --- a/docs/site/cli.md |
| +++ b/docs/site/cli.md |
| @@ -1561,9 +1561,15 @@ carlos economics backhaul --period mtd |
| ### carlos status |
| `--live` prints what every box in the fleet last reported about itself: |
| -release, tunnel state, memory, load, free disk. A box that has stopped |
| -reporting shows as UNREACHABLE with how long it has been dark, and none of |
| -its stale readings. |
| +release, tunnel state, memory, load, free disk, and whether its own alerts |
| +have anywhere to go. A box that has stopped reporting shows as UNREACHABLE |
| +with how long it has been dark, and none of its stale readings. |
| + |
| +The last column reads `alerts=sns`, `alerts=webhook` or `alerts=webhook+sns` |
| +when a box can deliver its alerts, and `alerts=not-delivered` when it has no |
| +sink configured. `alerts=sns-unavailable` means a topic is set but the box |
| +could not build its SNS client, so nothing is delivered through it. A dash |
| +means the box runs an older binary and did not say. |
| ```sh |
| carlos status --live |
diff --git a/go.mod b/go.mod| index f5e3fb69d..ae7c93327 100644 |
| --- a/go.mod |
| +++ b/go.mod |
| @@ -81,6 +81,7 @@ require ( |
| require ( |
| amadan.net/carlos/platform.git/launchctx v0.0.0 |
| + github.com/aws/aws-sdk-go-v2/service/sns v1.43.0 |
| github.com/oschwald/maxminddb-golang v1.13.1 |
| ) |
diff --git a/go.sum b/go.sum| index 17619db26..f81bc083a 100644 |
| --- a/go.sum |
| +++ b/go.sum |
| @@ -66,6 +66,8 @@ github.com/aws/aws-sdk-go-v2/service/sesv2 v1.66.6 h1:wc18zrr6hcDHI0uiwMxVXVebw7 |
| github.com/aws/aws-sdk-go-v2/service/sesv2 v1.66.6/go.mod h1:Ftv0JhdAfMkOhOQgZ7IEWkQh7MoOOxwfRhFUZgfZFtM= |
| github.com/aws/aws-sdk-go-v2/service/signin v1.0.4 h1:HpI7aMmJ+mm1wkSHIA2t5EaFFv5EFYXePW30p1EIrbQ= |
| github.com/aws/aws-sdk-go-v2/service/signin v1.0.4/go.mod h1:C5RdGMYGlfM0gYq/tifqgn4EbyX99V15P2V3R+VHbQU= |
| +github.com/aws/aws-sdk-go-v2/service/sns v1.43.0 h1:VPYjwn0BoX34hb44OT8T+Ikgn4NzsN7fHetaHaevsDc= |
| +github.com/aws/aws-sdk-go-v2/service/sns v1.43.0/go.mod h1:I1vnLPvvi9KBqxddu8nJ4vktoPJvaIG05UmjBD9sqm8= |
| github.com/aws/aws-sdk-go-v2/service/sso v1.30.8 h1:aM/Q24rIlS3bRAhTyFurowU8A0SMyGDtEOY/l/s/1Uw= |
| github.com/aws/aws-sdk-go-v2/service/sso v1.30.8/go.mod h1:+fWt2UHSb4kS7Pu8y+BMBvJF0EWx+4H0hzNwtDNRTrg= |
| github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 h1:AHDr0DaHIAo8c9t1emrzAlVDFp+iMMKnPdYy6XO4MCE= |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index f4eafc19f..91300c1cc 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -2791,6 +2791,10 @@ func statusLoop(reg *registry.Registry) { |
| // carlos-restart.path is a box where `carlos restart` answers |
| // "requested" and nothing cycles until the hourly pass (issue #113). |
| RestartPathRunning: restartPathActive(), |
| + // Whether this box's own alert rules have anywhere to deliver, |
| + // read from the env the metrics loop's sinks are built from |
| + // (discussion #192: an unset webhook was silent for weeks). |
| + Alerts: alertDeliveryProbe(os.Getenv, nil), |
| // ExpiredCanaries answers spec §5 for this box's own routes; see |
| // countExpiredCanaries and the canaryStore/canaryPub comment above. |
| ExpiredCanaries: countExpiredCanaries(ctx, routes, canaryStore, canaryPub, time.Now()), |
diff --git a/internal/host/alertsink.go b/internal/host/alertsink.go| new file mode 100644 |
| index 000000000..c9aca34d8 |
| --- /dev/null |
| +++ b/internal/host/alertsink.go |
| @@ -0,0 +1,336 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "math" |
| + "regexp" |
| + "strings" |
| + "sync/atomic" |
| + "time" |
| + |
| + "github.com/aws/aws-sdk-go-v2/aws/arn" |
| + awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| + "github.com/aws/aws-sdk-go-v2/service/sns" |
| + |
| + "github.com/carlosframework/platform/internal/metrics" |
| + "github.com/carlosframework/platform/internal/status" |
| +) |
| + |
| +// Alert sinks: where a firing or clearing transition from metrics.Evaluate |
| +// goes once alertTick has published the new state to the bucket. |
| +// |
| +// Two sinks, either, both, or neither: |
| +// |
| +// CARLOS_ALERT_WEBHOOK the event as JSON, POSTed (httpPostWebhook) |
| +// CARLOS_ALERT_SNS_TOPIC_ARN a plain-text subject and message, published |
| +// to that SNS topic with the box's instance role |
| +// |
| +// SNS is here because a webhook cannot reach it: Publish needs a SigV4 |
| +// signature, which only the AWS SDK on a box holding the role can make. The |
| +// flagship's topics (carlos-flagship-<region>-alerts) already carry confirmed |
| +// email subscriptions, so this is the path from a disk at 91% to an inbox |
| +// (discussion #192). |
| +// |
| +// Delivery is best effort and never fatal. A sink that fails is logged and |
| +// counted, the other sink still runs, and the tick carries on. Failed sends |
| +// are not rescheduled by a later tick: Evaluate emits an event only on a |
| +// transition, so the next tick has no event to resend. (The SDK itself tries a |
| +// retryable failure up to three times inside one publish, and a failed bucket |
| +// write in alertTick repeats the whole transition next tick, delivery |
| +// included.) The count is what makes a lost send visible, through |
| +// status.AlertDelivery in the box snapshot. |
| + |
| +// snsPublisher is the one SNS call this file makes. |
| +type snsPublisher interface { |
| + Publish(context.Context, *sns.PublishInput, ...func(*sns.Options)) (*sns.PublishOutput, error) |
| +} |
| + |
| +// alertPublishTimeout bounds one SNS publish, so a black-holed endpoint |
| +// cannot hold the 60 second sampler tick for the SDK's own default retries. |
| +const alertPublishTimeout = 10 * time.Second |
| + |
| +// alertCounters counts failed sends since the process started, and holds the |
| +// one fact the status loop cannot derive from the env: whether the SNS client |
| +// was actually built. They are separate goroutines, so this is their seam. |
| +type alertCounters struct { |
| + webhook atomic.Int64 |
| + sns atomic.Int64 |
| + // snsInit is snsInitUnknown until buildAlertSinks has run for a valid |
| + // topic ARN, then snsInitOK or snsInitFailed. |
| + snsInit atomic.Int32 |
| +} |
| + |
| +const ( |
| + snsInitUnknown int32 = iota |
| + snsInitOK |
| + snsInitFailed |
| +) |
| + |
| +// processAlertCounters is the process-wide count the metrics loop writes |
| +// and the status loop reads. They are separate goroutines with no other |
| +// shared state, so this is the seam between them. |
| +var processAlertCounters alertCounters |
| + |
| +// alertSinks is the configured delivery for one agent. |
| +type alertSinks struct { |
| + webhookURL string |
| + post func(url string, body []byte) error |
| + topicARN string |
| + pub snsPublisher |
| + counts *alertCounters |
| +} |
| + |
| +// newAlertSinks builds a sink set. An empty webhookURL or topicARN switches |
| +// that sink off; a nil publisher does the same for SNS. |
| +func newAlertSinks(webhookURL string, post func(url string, body []byte) error, |
| + topicARN string, pub snsPublisher, counts *alertCounters) *alertSinks { |
| + if counts == nil { |
| + counts = &processAlertCounters |
| + } |
| + return &alertSinks{webhookURL: webhookURL, post: post, topicARN: topicARN, pub: pub, counts: counts} |
| +} |
| + |
| +// deliver sends one transition to every configured sink. Safe on a nil |
| +// receiver, which means "no sinks". |
| +func (a *alertSinks) deliver(ctx context.Context, rules metrics.Rules, at time.Time, ev metrics.Event) { |
| + if a == nil { |
| + return |
| + } |
| + if a.webhookURL != "" && a.post != nil { |
| + evBody, err := json.Marshal(ev) |
| + if err != nil { |
| + log.Printf("host: alert event %s: marshal: %v", ev.Rule, err) |
| + } else if err := a.post(a.webhookURL, evBody); err != nil { |
| + a.counts.webhook.Add(1) |
| + log.Printf("host: alert webhook post (rule %s): %v", ev.Rule, err) |
| + } |
| + } |
| + if a.topicARN != "" && a.pub != nil { |
| + subject, message := formatAlert(rules, at, ev) |
| + pctx, cancel := context.WithTimeout(ctx, alertPublishTimeout) |
| + defer cancel() |
| + _, err := a.pub.Publish(pctx, &sns.PublishInput{ |
| + TopicArn: &a.topicARN, |
| + Subject: &subject, |
| + Message: &message, |
| + }) |
| + if err != nil { |
| + a.counts.sns.Add(1) |
| + log.Printf("host: alert sns publish (rule %s): %v", ev.Rule, err) |
| + } |
| + } |
| +} |
| + |
| +// topicARNPattern is the whole shape of a standard SNS topic ARN: partition, |
| +// region, a 12 digit account, and a topic name that is only letters, digits, |
| +// hyphen and underscore. The name has no colon, so a subscription ARN |
| +// (topic ARN plus ":<subscription-id>") does not match, and no dot, so FIFO |
| +// topics (which need a message group id this sink does not send) do not |
| +// either. |
| +var topicARNPattern = regexp.MustCompile(`^arn:aws(-cn|-us-gov)?:sns:([a-z0-9-]+):[0-9]{12}:[A-Za-z0-9_-]{1,256}$`) |
| + |
| +// parseAlertTopicARN checks CARLOS_ALERT_SNS_TOPIC_ARN and returns the |
| +// region the topic lives in. SNS is regional and a topic is reached only |
| +// through its own region's endpoint, so the region is read from the ARN and |
| +// not from AWS_REGION (host.env pins that to the deployment bucket's region |
| +// on every flagship box, which would aim three of the four at the wrong |
| +// place). |
| +func parseAlertTopicARN(s string) (region string, err error) { |
| + s = strings.TrimSpace(s) |
| + if s == "" { |
| + return "", errors.New("empty") |
| + } |
| + a, err := arn.Parse(s) |
| + if err != nil { |
| + return "", err |
| + } |
| + if a.Service != "sns" { |
| + return "", fmt.Errorf("service %q, want sns", a.Service) |
| + } |
| + if !topicARNPattern.MatchString(s) { |
| + return "", errors.New("not a topic arn (want arn:aws:sns:<region>:<12 digit account>:<topic name>)") |
| + } |
| + return a.Region, nil |
| +} |
| + |
| +// alertDeliveryProbe is the box's answer to "do my alert rules have anywhere |
| +// to go". The webhook and the ARN's shape come from the same env |
| +// buildAlertSinks reads; whether the SNS client was actually built comes from |
| +// counts, which buildAlertSinks fills in. So the snapshot reports the sink the |
| +// agent really holds: a topic ARN that does not parse is not a sink, and one |
| +// whose client failed to build reads SNS=false, SNSUnavailable=true. |
| +func alertDeliveryProbe(getenv func(string) string, counts *alertCounters) *status.AlertDelivery { |
| + if counts == nil { |
| + counts = &processAlertCounters |
| + } |
| + out := &status.AlertDelivery{ |
| + Webhook: strings.TrimSpace(getenv("CARLOS_ALERT_WEBHOOK")) != "", |
| + WebhookFailures: counts.webhook.Load(), |
| + SNSFailures: counts.sns.Load(), |
| + } |
| + if _, err := parseAlertTopicARN(getenv("CARLOS_ALERT_SNS_TOPIC_ARN")); err == nil { |
| + // Only an initialised client is a sink. snsInitUnknown (not built |
| + // yet, or the metrics loop never got there) is neither available |
| + // nor unavailable: the box has no working SNS sink to report. |
| + switch counts.snsInit.Load() { |
| + case snsInitOK: |
| + out.SNS = true |
| + case snsInitFailed: |
| + out.SNSUnavailable = true |
| + } |
| + } |
| + return out |
| +} |
| + |
| +// snsClientFactory builds the publisher for a region. It is a parameter so |
| +// tests can make client construction fail without an AWS config. |
| +type snsClientFactory func(ctx context.Context, region string) (snsPublisher, error) |
| + |
| +// newSNSClient is the production factory: the default credential chain (the |
| +// box's instance role), aimed at the topic's own region. |
| +func newSNSClient(ctx context.Context, region string) (snsPublisher, error) { |
| + cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region)) |
| + if err != nil { |
| + return nil, err |
| + } |
| + return sns.NewFromConfig(cfg), nil |
| +} |
| + |
| +// buildAlertSinks reads the sink env and builds the sinks. A bad topic ARN |
| +// is logged once and leaves SNS off; a client that cannot be built does the |
| +// same, and is also recorded in counts (state and one counted failure) so the |
| +// status snapshot reports the sink the agent really has. Neither stops the |
| +// metrics loop. With nothing configured it says so in the journal, because |
| +// that is the state that used to be silent. |
| +func buildAlertSinks(ctx context.Context, getenv func(string) string, counts *alertCounters, newClient snsClientFactory) *alertSinks { |
| + webhook := strings.TrimSpace(getenv("CARLOS_ALERT_WEBHOOK")) |
| + topic := strings.TrimSpace(getenv("CARLOS_ALERT_SNS_TOPIC_ARN")) |
| + |
| + var pub snsPublisher |
| + if topic != "" { |
| + region, err := parseAlertTopicARN(topic) |
| + if err != nil { |
| + log.Printf("agent: alerts: CARLOS_ALERT_SNS_TOPIC_ARN is not usable, SNS sink off: %v", err) |
| + topic = "" |
| + } else if p, err := newClient(ctx, region); err != nil { |
| + log.Printf("agent: alerts: build SNS client, SNS sink off: %v", err) |
| + counts.sns.Add(1) |
| + counts.snsInit.Store(snsInitFailed) |
| + topic = "" |
| + } else { |
| + counts.snsInit.Store(snsInitOK) |
| + pub = p |
| + } |
| + } |
| + |
| + switch { |
| + case webhook == "" && topic == "": |
| + log.Printf("agent: alerts: NO SINK CONFIGURED (set CARLOS_ALERT_SNS_TOPIC_ARN or CARLOS_ALERT_WEBHOOK); disk, memory and cpu alerts will be published to the bucket and delivered nowhere") |
| + case webhook != "" && topic != "": |
| + log.Printf("agent: alerts: delivering to webhook and SNS topic %s", topic) |
| + case topic != "": |
| + log.Printf("agent: alerts: delivering to SNS topic %s", topic) |
| + default: |
| + log.Printf("agent: alerts: delivering to webhook") |
| + } |
| + return newAlertSinks(webhook, httpPostWebhook, topic, pub, counts) |
| +} |
| + |
| +// formatAlert renders one transition as an email: a one-line subject and a |
| +// short plain-text body. The inputs are the event, the rule levels and a |
| +// time, and nothing else, so no URL, ARN or credential can reach the |
| +// message. The subject is ASCII with no newline and under 100 characters, |
| +// which is what SNS accepts. |
| +func formatAlert(rules metrics.Rules, at time.Time, ev metrics.Event) (subject, body string) { |
| + subjectLine := alertSubject(ev) |
| + var b strings.Builder |
| + if ev.Firing { |
| + fmt.Fprintf(&b, "%s.\n\n", alertSentence(ev)) |
| + } else { |
| + fmt.Fprintf(&b, "%s.\n\n", alertClearSentence(ev)) |
| + } |
| + if level, ok := alertLevel(rules, ev.Rule); ok { |
| + fmt.Fprintf(&b, "the alert level is %s.\n", level) |
| + } |
| + if ev.Firing { |
| + if d := alertSustain(rules, ev.Rule); d > 0 { |
| + fmt.Fprintf(&b, "it stayed above that level for %d minutes before this alert was raised.\n", int(d.Minutes())) |
| + } |
| + } |
| + fmt.Fprintf(&b, "this was reported at %s.\n", at.UTC().Format("2006-01-02 15:04 MST")) |
| + if ev.Firing { |
| + b.WriteString("you will get another email when it clears.\n") |
| + } |
| + fmt.Fprintf(&b, "\nrule: %s\nbox: %s\n", ev.Rule, ev.Box) |
| + return subjectLine, b.String() |
| +} |
| + |
| +func alertSubject(ev metrics.Event) string { |
| + if ev.Firing { |
| + return "carlos alert: " + alertSentence(ev) |
| + } |
| + return "carlos alert cleared: " + alertClearSentence(ev) |
| +} |
| + |
| +// alertSentence is the lower-case "what is wrong" clause, shared by the |
| +// subject and the body. |
| +func alertSentence(ev metrics.Event) string { |
| + switch ev.Rule { |
| + case "disk": |
| + return fmt.Sprintf("disk is %s full on %s", pct(ev.Value), ev.Box) |
| + case "mem": |
| + return fmt.Sprintf("memory use is %s on %s", pct(ev.Value), ev.Box) |
| + case "cpu": |
| + return fmt.Sprintf("cpu use is %s on %s", pct(ev.Value), ev.Box) |
| + case "lag": |
| + return fmt.Sprintf("mail relay lag is %d seconds on %s", int64(ev.Value), ev.Box) |
| + default: |
| + return fmt.Sprintf("%s is firing on %s", ev.Rule, ev.Box) |
| + } |
| +} |
| + |
| +func alertClearSentence(ev metrics.Event) string { |
| + switch ev.Rule { |
| + case "disk": |
| + return fmt.Sprintf("disk is back to %s full on %s", pct(ev.Value), ev.Box) |
| + case "mem": |
| + return fmt.Sprintf("memory use is back to %s on %s", pct(ev.Value), ev.Box) |
| + case "cpu": |
| + return fmt.Sprintf("cpu use is back to %s on %s", pct(ev.Value), ev.Box) |
| + case "lag": |
| + return fmt.Sprintf("mail relay lag is back to %d seconds on %s", int64(ev.Value), ev.Box) |
| + default: |
| + return fmt.Sprintf("%s has cleared on %s", ev.Rule, ev.Box) |
| + } |
| +} |
| + |
| +func alertLevel(r metrics.Rules, rule string) (string, bool) { |
| + switch rule { |
| + case "disk": |
| + return pct(r.DiskFrac), true |
| + case "mem": |
| + return pct(r.MemFrac), true |
| + case "cpu": |
| + return pct(r.CPUFrac), true |
| + case "lag": |
| + return fmt.Sprintf("%d seconds", r.LagSecs), true |
| + } |
| + return "", false |
| +} |
| + |
| +func alertSustain(r metrics.Rules, rule string) time.Duration { |
| + switch rule { |
| + case "mem": |
| + return r.MemSustain |
| + case "cpu": |
| + return r.CPUSustain |
| + } |
| + return 0 |
| +} |
| + |
| +// pct renders a 0..1 fraction as a whole percent. |
| +func pct(f float64) string { return fmt.Sprintf("%d%%", int(math.Round(f*100))) } |
diff --git a/internal/host/alertsink_test.go b/internal/host/alertsink_test.go| new file mode 100644 |
| index 000000000..bd162e674 |
| --- /dev/null |
| +++ b/internal/host/alertsink_test.go |
| @@ -0,0 +1,451 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/aws/aws-sdk-go-v2/service/sns" |
| + |
| + "github.com/carlosframework/platform/internal/metrics" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// fakeSNS records every Publish and can be told to fail. It is the only SNS |
| +// the tests in this file ever talk to. |
| +type fakeSNS struct { |
| + in []*sns.PublishInput |
| + err error |
| +} |
| + |
| +func (f *fakeSNS) Publish(_ context.Context, in *sns.PublishInput, _ ...func(*sns.Options)) (*sns.PublishOutput, error) { |
| + f.in = append(f.in, in) |
| + if f.err != nil { |
| + return nil, f.err |
| + } |
| + return &sns.PublishOutput{}, nil |
| +} |
| + |
| +const testTopicARN = "arn:aws:sns:eu-west-1:271376211898:carlos-flagship-ie-alerts" |
| + |
| +var sinkTestAt = time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC) |
| + |
| +func TestFormatAlertFireAndClear(t *testing.T) { |
| + rules := metrics.DefaultRules() |
| + cases := []struct { |
| + name string |
| + ev metrics.Event |
| + subjectHas []string |
| + bodyHas []string |
| + subjectNot []string |
| + bodyNotHave []string |
| + }{ |
| + { |
| + name: "disk fires", |
| + ev: metrics.Event{Rule: "disk", Firing: true, Value: 0.917, Box: "ada-stargazing"}, |
| + subjectHas: []string{"carlos alert", "disk", "92%", "ada-stargazing"}, |
| + bodyHas: []string{"disk", "92%", "ada-stargazing", "90%", "2026-10-05 12:00 UTC", "clears"}, |
| + }, |
| + { |
| + name: "disk clears", |
| + ev: metrics.Event{Rule: "disk", Firing: false, Value: 0.30, Box: "ada-stargazing"}, |
| + subjectHas: []string{"carlos alert cleared", "disk", "30%", "ada-stargazing"}, |
| + bodyHas: []string{"back", "30%", "ada-stargazing"}, |
| + }, |
| + { |
| + name: "mem fires with its sustain window", |
| + ev: metrics.Event{Rule: "mem", Firing: true, Value: 0.94, Box: "box-a"}, |
| + subjectHas: []string{"memory", "94%", "box-a"}, |
| + bodyHas: []string{"10 minutes", "90%"}, |
| + }, |
| + { |
| + name: "cpu fires with its sustain window", |
| + ev: metrics.Event{Rule: "cpu", Firing: true, Value: 0.97, Box: "box-a"}, |
| + subjectHas: []string{"cpu", "97%", "box-a"}, |
| + bodyHas: []string{"15 minutes", "95%"}, |
| + }, |
| + { |
| + name: "lag is seconds not a percentage", |
| + ev: metrics.Event{Rule: "lag", Firing: true, Value: 1200, Box: "box-a"}, |
| + subjectHas: []string{"lag", "1200 seconds"}, |
| + subjectNot: []string{"%"}, |
| + bodyHas: []string{"900 seconds"}, |
| + }, |
| + { |
| + name: "an unknown rule still reads", |
| + ev: metrics.Event{Rule: "newrule", Firing: true, Value: 0.5, Box: "box-a"}, |
| + subjectHas: []string{"newrule", "box-a"}, |
| + }, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.name, func(t *testing.T) { |
| + subject, body := formatAlert(rules, sinkTestAt, c.ev) |
| + for _, w := range c.subjectHas { |
| + if !strings.Contains(subject, w) { |
| + t.Errorf("subject %q missing %q", subject, w) |
| + } |
| + } |
| + for _, w := range c.subjectNot { |
| + if strings.Contains(subject, w) { |
| + t.Errorf("subject %q should not contain %q", subject, w) |
| + } |
| + } |
| + for _, w := range c.bodyHas { |
| + if !strings.Contains(body, w) { |
| + t.Errorf("body missing %q:\n%s", w, body) |
| + } |
| + } |
| + // SNS rejects a subject over 100 chars, or with a newline or |
| + // any non-printable-ASCII byte; the mail would never be sent. |
| + if len(subject) == 0 || len(subject) > 100 { |
| + t.Errorf("subject length %d, want 1..100: %q", len(subject), subject) |
| + } |
| + for _, r := range subject { |
| + if r < 0x20 || r > 0x7e { |
| + t.Errorf("subject has a character SNS refuses: %q", subject) |
| + break |
| + } |
| + } |
| + for _, s := range []string{subject, body} { |
| + if strings.ContainsRune(s, '\u2014') || strings.ContainsRune(s, '\u2013') { |
| + t.Errorf("em or en dash in copy: %q", s) |
| + } |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// TestFormatAlertCarriesNoSecrets: the message is the event, the rule's |
| +// level and a time. No URL, ARN, credential, or env value can reach it, |
| +// because nothing but those three inputs is passed in. |
| +func TestFormatAlertCarriesNoSecrets(t *testing.T) { |
| + subject, body := formatAlert(metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "disk", Firing: true, Value: 0.95, Box: "box-a"}) |
| + for _, banned := range []string{"http", "arn:", "secret", "token", "key", "password"} { |
| + if strings.Contains(strings.ToLower(subject+body), banned) { |
| + t.Errorf("message contains %q:\n%s\n%s", banned, subject, body) |
| + } |
| + } |
| +} |
| + |
| +func TestAlertSinksDeliverSNSOnly(t *testing.T) { |
| + pub := &fakeSNS{} |
| + counts := &alertCounters{} |
| + s := newAlertSinks("", nil, testTopicARN, pub, counts) |
| + s.deliver(context.Background(), metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "disk", Firing: true, Value: 0.95, Box: "box-a"}) |
| + if len(pub.in) != 1 { |
| + t.Fatalf("publishes = %d, want 1", len(pub.in)) |
| + } |
| + in := pub.in[0] |
| + if got := deref(in.TopicArn); got != testTopicARN { |
| + t.Errorf("TopicArn = %q", got) |
| + } |
| + if !strings.Contains(deref(in.Subject), "disk") || !strings.Contains(deref(in.Message), "box-a") { |
| + t.Errorf("subject/message not readable: %q / %q", deref(in.Subject), deref(in.Message)) |
| + } |
| + if counts.sns.Load() != 0 || counts.webhook.Load() != 0 { |
| + t.Errorf("no failures expected, got sns=%d webhook=%d", counts.sns.Load(), counts.webhook.Load()) |
| + } |
| +} |
| + |
| +func TestAlertSinksDeliverWebhookOnlyIsUnchanged(t *testing.T) { |
| + var posts [][]byte |
| + var urls []string |
| + post := func(url string, body []byte) error { |
| + urls = append(urls, url) |
| + posts = append(posts, body) |
| + return nil |
| + } |
| + s := newAlertSinks("http://example.test/hook", post, "", nil, &alertCounters{}) |
| + s.deliver(context.Background(), metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "disk", Firing: true, Value: 0.95, Box: "box-a"}) |
| + if len(posts) != 1 || urls[0] != "http://example.test/hook" { |
| + t.Fatalf("posts = %d urls = %v", len(posts), urls) |
| + } |
| + if string(posts[0]) != `{"rule":"disk","firing":true,"value":0.95,"box":"box-a"}` { |
| + t.Errorf("webhook payload changed: %s", posts[0]) |
| + } |
| +} |
| + |
| +func TestAlertSinksBothAreUsed(t *testing.T) { |
| + pub := &fakeSNS{} |
| + posts := 0 |
| + post := func(string, []byte) error { posts++; return nil } |
| + s := newAlertSinks("http://example.test/hook", post, testTopicARN, pub, &alertCounters{}) |
| + s.deliver(context.Background(), metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "mem", Firing: true, Value: 0.95, Box: "box-a"}) |
| + if posts != 1 || len(pub.in) != 1 { |
| + t.Errorf("webhook posts = %d, sns publishes = %d, want 1 and 1", posts, len(pub.in)) |
| + } |
| +} |
| + |
| +func TestAlertSinksNoneConfiguredDoesNothing(t *testing.T) { |
| + s := newAlertSinks("", nil, "", nil, &alertCounters{}) |
| + s.deliver(context.Background(), metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "disk", Firing: true, Value: 0.95, Box: "box-a"}) |
| + var nilSinks *alertSinks |
| + nilSinks.deliver(context.Background(), metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "disk", Firing: true, Value: 0.95, Box: "box-a"}) |
| +} |
| + |
| +// TestAlertSinksSNSFailureIsCountedAndNeverBlocksTheWebhook: a failing sink |
| +// is logged and counted, the other sink still runs, and nothing panics or |
| +// returns an error that could fail the tick. |
| +func TestAlertSinksSNSFailureIsCountedAndNeverBlocksTheWebhook(t *testing.T) { |
| + pub := &fakeSNS{err: errors.New("AccessDenied: not authorized to perform sns:Publish")} |
| + posts := 0 |
| + post := func(string, []byte) error { posts++; return nil } |
| + counts := &alertCounters{} |
| + s := newAlertSinks("http://example.test/hook", post, testTopicARN, pub, counts) |
| + s.deliver(context.Background(), metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "disk", Firing: true, Value: 0.95, Box: "box-a"}) |
| + if posts != 1 { |
| + t.Errorf("webhook posts = %d, want 1 even though sns failed", posts) |
| + } |
| + if counts.sns.Load() != 1 { |
| + t.Errorf("sns failures = %d, want 1", counts.sns.Load()) |
| + } |
| + if counts.webhook.Load() != 0 { |
| + t.Errorf("webhook failures = %d, want 0", counts.webhook.Load()) |
| + } |
| +} |
| + |
| +func TestAlertSinksWebhookFailureIsCounted(t *testing.T) { |
| + pub := &fakeSNS{} |
| + post := func(string, []byte) error { return errors.New("boom") } |
| + counts := &alertCounters{} |
| + s := newAlertSinks("http://example.test/hook", post, testTopicARN, pub, counts) |
| + s.deliver(context.Background(), metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "disk", Firing: true, Value: 0.95, Box: "box-a"}) |
| + if counts.webhook.Load() != 1 || len(pub.in) != 1 { |
| + t.Errorf("webhook failures = %d (want 1), sns publishes = %d (want 1)", counts.webhook.Load(), len(pub.in)) |
| + } |
| +} |
| + |
| +// TestAlertTickDeliversThroughSinks: alertTick hands each transition to the |
| +// sinks exactly once, and a steady state sends nothing more. |
| +func TestAlertTickDeliversThroughSinks(t *testing.T) { |
| + ctx := context.Background() |
| + store := release.NewDirStore(t.TempDir()) |
| + statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json") |
| + box := "box-a" |
| + rules := metrics.DefaultRules() |
| + st := &metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}} |
| + pub := &fakeSNS{} |
| + sinks := newAlertSinks("", nil, testTopicARN, pub, &alertCounters{}) |
| + |
| + disk := 0.95 |
| + latest := metrics.Sample{At: sinkTestAt, Host: metrics.HostSample{DiskFrac: &disk}} |
| + if err := alertTick(ctx, store, box, rules, st, latest, statePath, sinks); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := alertTick(ctx, store, box, rules, st, latest, statePath, sinks); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(pub.in) != 1 { |
| + t.Fatalf("publishes = %d, want 1 (second tick has no transition)", len(pub.in)) |
| + } |
| + if !strings.Contains(deref(pub.in[0].Message), box) { |
| + t.Errorf("message does not name the box: %q", deref(pub.in[0].Message)) |
| + } |
| + |
| + // Clearing is a transition too. |
| + low := 0.30 |
| + cleared := metrics.Sample{At: sinkTestAt.Add(time.Minute), Host: metrics.HostSample{DiskFrac: &low}} |
| + if err := alertTick(ctx, store, box, rules, st, cleared, statePath, sinks); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(pub.in) != 2 || !strings.Contains(deref(pub.in[1].Subject), "cleared") { |
| + t.Errorf("clear not published: %d publishes", len(pub.in)) |
| + } |
| + if _, err := os.Stat(statePath); err != nil { |
| + t.Errorf("state not persisted: %v", err) |
| + } |
| +} |
| + |
| +func TestParseAlertTopicARN(t *testing.T) { |
| + cases := []struct { |
| + in string |
| + wantRegion string |
| + wantErr bool |
| + }{ |
| + {testTopicARN, "eu-west-1", false}, |
| + {"arn:aws:sns:ap-southeast-2:271376211898:carlos-flagship-au-alerts", "ap-southeast-2", false}, |
| + {" " + testTopicARN + "\n", "eu-west-1", false}, |
| + {"", "", true}, |
| + {"not-an-arn", "", true}, |
| + {"arn:aws:s3:::some-bucket", "", true}, |
| + {"arn:aws:sqs:eu-west-1:271376211898:queue", "", true}, |
| + {"arn:aws:sns:::topic", "", true}, |
| + {"arn:aws:sns:eu-west-1:271376211898:", "", true}, |
| + // A subscription ARN is topic ARN plus ":<subscription-id>". It is |
| + // an sns ARN with all the right fields, and publishing to it fails. |
| + {testTopicARN + ":8f4c0e7a-1b2c-4d3e-9f50-0123456789ab", "", true}, |
| + {"arn:aws:sns:eu-west-1:271376211898:topic:subscription-id", "", true}, |
| + {"arn:aws:sns:eu-west-1:271376211898:a:b", "", true}, |
| + {"arn:aws:sns:eu-west-1:271376211898:name.fifo", "", true}, |
| + {"arn:aws:sns:eu-west-1:271376211898:bad name", "", true}, |
| + {"arn:aws:sns:eu-west-1:27137621189:carlos-alerts", "", true}, |
| + {"arn:aws:sns:eu-west-1:abcdefghijkl:carlos-alerts", "", true}, |
| + {"arn:aws:sns:eu-west-1:271376211898:carlos-alerts\nx", "", true}, |
| + {"arn:aws:sns:eu-west-1:271376211898:*", "", true}, |
| + } |
| + for _, c := range cases { |
| + region, err := parseAlertTopicARN(c.in) |
| + if (err != nil) != c.wantErr || region != c.wantRegion { |
| + t.Errorf("parseAlertTopicARN(%q) = %q, %v; want %q, err=%v", c.in, region, err, c.wantRegion, c.wantErr) |
| + } |
| + } |
| +} |
| + |
| +func TestAlertDeliveryProbe(t *testing.T) { |
| + env := func(m map[string]string) func(string) string { return func(k string) string { return m[k] } } |
| + |
| + counts := &alertCounters{} |
| + got := alertDeliveryProbe(env(nil), counts) |
| + if got.Webhook || got.SNS || got.Delivered() { |
| + t.Errorf("nothing set: %+v, want nothing delivered", got) |
| + } |
| + |
| + got = alertDeliveryProbe(env(map[string]string{"CARLOS_ALERT_WEBHOOK": "http://x"}), counts) |
| + if !got.Webhook || got.SNS || !got.Delivered() { |
| + t.Errorf("webhook only: %+v", got) |
| + } |
| + |
| + // A valid ARN alone is not a sink until buildAlertSinks has built the |
| + // client: before that (a stalled build, or a metrics loop that exited |
| + // first) the box must not claim sns. |
| + got = alertDeliveryProbe(env(map[string]string{"CARLOS_ALERT_SNS_TOPIC_ARN": testTopicARN}), counts) |
| + if got.SNS || got.SNSUnavailable || got.Delivered() { |
| + t.Errorf("sink not yet built must not read as sns: %+v", got) |
| + } |
| + |
| + counts.snsInit.Store(snsInitOK) |
| + got = alertDeliveryProbe(env(map[string]string{"CARLOS_ALERT_SNS_TOPIC_ARN": testTopicARN}), counts) |
| + if got.Webhook || !got.SNS || !got.Delivered() { |
| + t.Errorf("sns only: %+v", got) |
| + } |
| + |
| + // A malformed ARN is not a sink: the box has nowhere to deliver, and |
| + // the snapshot must say so rather than claim an sns sink it cannot use. |
| + got = alertDeliveryProbe(env(map[string]string{"CARLOS_ALERT_SNS_TOPIC_ARN": "nonsense"}), counts) |
| + if got.SNS || got.Delivered() { |
| + t.Errorf("bad arn counted as a sink: %+v", got) |
| + } |
| + |
| + counts.sns.Add(3) |
| + counts.webhook.Add(2) |
| + got = alertDeliveryProbe(env(map[string]string{"CARLOS_ALERT_SNS_TOPIC_ARN": testTopicARN}), counts) |
| + if got.SNSFailures != 3 || got.WebhookFailures != 2 { |
| + t.Errorf("failure counts not carried: %+v", got) |
| + } |
| +} |
| + |
| +// fakeClientFactory is the buildAlertSinks seam: it records the region it |
| +// was asked for and either hands back a fake or fails. |
| +type fakeClientFactory struct { |
| + regions []string |
| + err error |
| + pub *fakeSNS |
| +} |
| + |
| +func (f *fakeClientFactory) build(_ context.Context, region string) (snsPublisher, error) { |
| + f.regions = append(f.regions, region) |
| + if f.err != nil { |
| + return nil, f.err |
| + } |
| + return f.pub, nil |
| +} |
| + |
| +func TestBuildAlertSinksInitialisedSNS(t *testing.T) { |
| + env := func(m map[string]string) func(string) string { return func(k string) string { return m[k] } } |
| + counts := &alertCounters{} |
| + f := &fakeClientFactory{pub: &fakeSNS{}} |
| + sinks := buildAlertSinks(context.Background(), env(map[string]string{"CARLOS_ALERT_SNS_TOPIC_ARN": testTopicARN}), counts, f.build) |
| + if len(f.regions) != 1 || f.regions[0] != "eu-west-1" { |
| + t.Fatalf("client built for %v, want the ARN's region eu-west-1", f.regions) |
| + } |
| + sinks.deliver(context.Background(), metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "disk", Firing: true, Value: 0.95, Box: "box-a"}) |
| + if len(f.pub.in) != 1 { |
| + t.Fatalf("publishes = %d, want 1", len(f.pub.in)) |
| + } |
| + got := alertDeliveryProbe(env(map[string]string{"CARLOS_ALERT_SNS_TOPIC_ARN": testTopicARN}), counts) |
| + if !got.SNS || got.SNSUnavailable || got.SNSFailures != 0 { |
| + t.Errorf("a working sink must read as sns: %+v", got) |
| + } |
| +} |
| + |
| +// TestBuildAlertSinksClientFailureIsVisible is the review's finding: with a |
| +// valid ARN in the env but a client that cannot be built (no region, bad |
| +// profile), the sink is off. The status probe must say so, not report the |
| +// env as if it were a working sink with zero failures. |
| +func TestBuildAlertSinksClientFailureIsVisible(t *testing.T) { |
| + envMap := map[string]string{"CARLOS_ALERT_SNS_TOPIC_ARN": testTopicARN} |
| + env := func(k string) string { return envMap[k] } |
| + counts := &alertCounters{} |
| + f := &fakeClientFactory{err: errors.New("failed to get shared config profile")} |
| + sinks := buildAlertSinks(context.Background(), env, counts, f.build) |
| + // Delivering must neither panic nor publish. |
| + sinks.deliver(context.Background(), metrics.DefaultRules(), sinkTestAt, |
| + metrics.Event{Rule: "disk", Firing: true, Value: 0.95, Box: "box-a"}) |
| + |
| + got := alertDeliveryProbe(env, counts) |
| + if got.SNS || !got.SNSUnavailable || got.Delivered() { |
| + t.Errorf("failed init must read as unavailable and not delivered: %+v", got) |
| + } |
| + if got.SNSFailures < 1 { |
| + t.Errorf("failed init must be counted, got %d", got.SNSFailures) |
| + } |
| + |
| + // With a webhook as well, the box still delivers, and says what is wrong. |
| + envMap["CARLOS_ALERT_WEBHOOK"] = "http://example.test/hook" |
| + got = alertDeliveryProbe(env, counts) |
| + if !got.Webhook || got.SNS || !got.SNSUnavailable || !got.Delivered() { |
| + t.Errorf("webhook alone must still deliver: %+v", got) |
| + } |
| +} |
| + |
| +func TestBuildAlertSinksBadARNIsNotASinkAndBuildsNoClient(t *testing.T) { |
| + counts := &alertCounters{} |
| + f := &fakeClientFactory{pub: &fakeSNS{}} |
| + env := func(k string) string { |
| + if k == "CARLOS_ALERT_SNS_TOPIC_ARN" { |
| + return testTopicARN + ":subscription-id" |
| + } |
| + return "" |
| + } |
| + buildAlertSinks(context.Background(), env, counts, f.build) |
| + if len(f.regions) != 0 { |
| + t.Errorf("a subscription ARN must not build a client: %v", f.regions) |
| + } |
| + got := alertDeliveryProbe(env, counts) |
| + if got.SNS || got.Delivered() { |
| + t.Errorf("subscription ARN counted as a sink: %+v", got) |
| + } |
| +} |
| + |
| +func TestBuildAlertSinksNothingConfigured(t *testing.T) { |
| + counts := &alertCounters{} |
| + f := &fakeClientFactory{pub: &fakeSNS{}} |
| + buildAlertSinks(context.Background(), func(string) string { return "" }, counts, f.build) |
| + if len(f.regions) != 0 { |
| + t.Errorf("no ARN set, yet a client was built: %v", f.regions) |
| + } |
| + got := alertDeliveryProbe(func(string) string { return "" }, counts) |
| + if got.Delivered() || got.SNSUnavailable { |
| + t.Errorf("nothing configured: %+v", got) |
| + } |
| +} |
| + |
| +func deref(s *string) string { |
| + if s == nil { |
| + return "" |
| + } |
| + return *s |
| +} |
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go| index fa4d9e44b..cbd2cc382 100644 |
| --- a/internal/host/metrics_loop.go |
| +++ b/internal/host/metrics_loop.go |
| @@ -442,7 +442,8 @@ func pruneStaleRaw(ctx context.Context, s release.Store, box string, now time.Ti |
| // alertTick evaluates rules against latest and, when that produces a |
| // firing/clearing transition (Evaluate emits Events exactly on those — see |
| // its own doc comment), publishes the new state to metrics.AlertKey(box) |
| -// and posts each event to webhookURL ("" skips the webhook). |
| +// and delivers each event to sinks (alertsink.go: the webhook, the SNS topic, |
| +// either, both or neither; nil is none). |
| // |
| // Retry semantics on a publish failure are the crux of this function: *st |
| // is the caller's durable record of "what has been reliably told to the |
| @@ -459,10 +460,9 @@ func pruneStaleRaw(ctx context.Context, s release.Store, box string, now time.Ti |
| // once *st already says "firing", every later tick just confirms it's |
| // still firing with no new event and nothing left to retry. |
| // |
| -// The webhook post is still attempted this tick regardless of whether |
| -// publish succeeded — it's independently fire-and-forget, logged and |
| -// otherwise ignored on failure, same as before; it does not gate whether |
| -// *st advances. Local persistence to statePath is also unconditionally |
| +// Delivery to the sinks is still attempted this tick regardless of whether |
| +// publish succeeded — it's independently fire-and-forget, logged and counted |
| +// on failure, never returned; it does not gate whether *st advances. Local persistence to statePath is also unconditionally |
| // best-effort (a failure is logged, never returned or allowed to block |
| // publish/webhook — see round 1's fix, since the local write is most likely |
| // to fail exactly when the disk rule is the one firing), but it always |
| @@ -483,8 +483,8 @@ func pruneStaleRaw(ctx context.Context, s release.Store, box string, now time.Ti |
| // rules (disk, lag; sustain 0) retry the same way, just starting from |
| // their very first over-threshold tick instead of a sustain deadline. |
| func alertTick(ctx context.Context, s release.Store, box string, rules metrics.Rules, |
| - st *metrics.AlertState, latest metrics.Sample, statePath, webhookURL string, |
| - post func(url string, body []byte) error) error { |
| + st *metrics.AlertState, latest metrics.Sample, statePath string, |
| + sinks *alertSinks) error { |
| obs := metrics.Observation{ |
| At: latest.At, |
| @@ -506,19 +506,9 @@ func alertTick(ctx context.Context, s release.Store, box string, rules metrics.R |
| } |
| // Fire-and-forget, independent of publish's outcome: an event still |
| - // deserves a webhook attempt even if the store write for it failed. |
| + // deserves a delivery attempt even if the store write for it failed. |
| for _, ev := range events { |
| - if webhookURL == "" { |
| - continue |
| - } |
| - evBody, err := json.Marshal(ev) |
| - if err != nil { |
| - log.Printf("host: alert event %s: marshal: %v", ev.Rule, err) |
| - continue |
| - } |
| - if err := post(webhookURL, evBody); err != nil { |
| - log.Printf("host: alert webhook post (rule %s): %v", ev.Rule, err) |
| - } |
| + sinks.deliver(ctx, rules, latest.At, ev) |
| } |
| committed := *st |
| @@ -811,7 +801,7 @@ func metricsLoop(execRoot string, stampedUnits func() []string) { |
| st := loadAlertState(statePath, box) |
| rules := metrics.RulesFromEnv(os.Getenv) |
| logRejectedAlertOverrides(os.Getenv) |
| - webhookURL := os.Getenv("CARLOS_ALERT_WEBHOOK") |
| + sinks := buildAlertSinks(ctx, os.Getenv, &processAlertCounters, newSNSClient) |
| log.Printf("agent: metrics loop armed (60s ticks, 5min flush), publishing to %s", target) |
| @@ -824,7 +814,7 @@ func metricsLoop(execRoot string, stampedUnits func() []string) { |
| sm.tick(probes) |
| latest := sm.buf[len(sm.buf)-1] |
| - if err := alertTick(ctx, s, box, rules, &st, latest, statePath, webhookURL, httpPostWebhook); err != nil { |
| + if err := alertTick(ctx, s, box, rules, &st, latest, statePath, sinks); err != nil { |
| log.Printf("agent: alert tick: %v", err) |
| } |
diff --git a/internal/host/metrics_loop_test.go b/internal/host/metrics_loop_test.go| index 7ad672f3a..6ce15716a 100644 |
| --- a/internal/host/metrics_loop_test.go |
| +++ b/internal/host/metrics_loop_test.go |
| @@ -143,8 +143,7 @@ func TestMetricsLoopStoreHonorsFleetPrefix(t *testing.T) { |
| firing := 0.95 |
| latest := metrics.Sample{At: at, Host: metrics.HostSample{DiskFrac: &firing}} |
| statePath := filepath.Join(t.TempDir(), "monitor", "alerts.json") |
| - noPost := func(string, []byte) error { return nil } |
| - if err := alertTick(ctx, s, box, rules, st, latest, statePath, "", noPost); err != nil { |
| + if err := alertTick(ctx, s, box, rules, st, latest, statePath, nil); err != nil { |
| t.Fatal(err) |
| } |
| if _, _, err := under.Get(ctx, "fleets/pi/"+metrics.AlertKey(box)); err != nil { |
| @@ -223,7 +222,7 @@ func TestAlertTickPersistsPublishesAndPosts(t *testing.T) { |
| return nil |
| } |
| - if err := alertTick(ctx, store, box, rules, st, latest, statePath, "http://example.test/webhook", post); err != nil { |
| + if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil { |
| t.Fatal(err) |
| } |
| @@ -249,7 +248,7 @@ func TestAlertTickPersistsPublishesAndPosts(t *testing.T) { |
| } |
| // Second, identical tick: already firing, no new transition -> no post. |
| - if err := alertTick(ctx, store, box, rules, st, latest, statePath, "http://example.test/webhook", post); err != nil { |
| + if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil { |
| t.Fatal(err) |
| } |
| if len(posts) != 1 { |
| @@ -275,7 +274,7 @@ func TestAlertTickWebhookFailureDoesNotBlock(t *testing.T) { |
| } |
| post := func(url string, body []byte) error { return errors.New("webhook unreachable") } |
| - if err := alertTick(ctx, store, box, rules, st, latest, statePath, "http://example.test/webhook", post); err != nil { |
| + if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil { |
| t.Fatalf("alertTick must not fail on a webhook error: %v", err) |
| } |
| if _, err := os.Stat(statePath); err != nil { |
| @@ -317,7 +316,7 @@ func TestAlertTickPersistFailureDoesNotSuppressPublish(t *testing.T) { |
| return nil |
| } |
| - if err := alertTick(ctx, store, box, rules, st, latest, statePath, "http://example.test/webhook", post); err != nil { |
| + if err := alertTick(ctx, store, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil { |
| t.Fatalf("alertTick must not fail just because the local state write failed: %v", err) |
| } |
| if !posted { |
| @@ -551,7 +550,7 @@ func TestAlertTickRetriesTransitionAfterPublishFailure(t *testing.T) { |
| } |
| failStore := &failingPutStore{Store: release.NewDirStore(t.TempDir())} |
| - if err := alertTick(ctx, failStore, box, rules, st, latest, statePath, "http://example.test/webhook", post); err == nil { |
| + if err := alertTick(ctx, failStore, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err == nil { |
| t.Fatal("expected alertTick to return the publish error against a Put-always-fails store") |
| } |
| if st.Rules["disk"].Firing { |
| @@ -564,7 +563,7 @@ func TestAlertTickRetriesTransitionAfterPublishFailure(t *testing.T) { |
| // Retry with the SAME observation against a working store: the pending |
| // transition must reproduce, publish, and post again. |
| workingStore := release.NewDirStore(t.TempDir()) |
| - if err := alertTick(ctx, workingStore, box, rules, st, latest, statePath, "http://example.test/webhook", post); err != nil { |
| + if err := alertTick(ctx, workingStore, box, rules, st, latest, statePath, webhookSinks("http://example.test/webhook", post)); err != nil { |
| t.Fatalf("retry against a working store: %v", err) |
| } |
| if !st.Rules["disk"].Firing { |
| @@ -918,3 +917,9 @@ func TestMetricsBoxLabelUsesFleetIdentity(t *testing.T) { |
| } |
| }) |
| } |
| + |
| +// webhookSinks is the sink set the pre-SNS tests meant: a webhook and |
| +// nothing else. |
| +func webhookSinks(url string, post func(url string, body []byte) error) *alertSinks { |
| + return newAlertSinks(url, post, "", nil, &alertCounters{}) |
| +} |
diff --git a/internal/status/snapshot.go b/internal/status/snapshot.go| index dd757d7dd..398c399d9 100644 |
| --- a/internal/status/snapshot.go |
| +++ b/internal/status/snapshot.go |
| @@ -80,6 +80,9 @@ type Probes struct { |
| // zero; 0 is itself a real, confident answer when every canary route on |
| // the box is fresh, or there are none at all. |
| ExpiredCanaries *int |
| + // Alerts is the box's alert-sink answer (see AlertDelivery), carried |
| + // into Box.Metrics. nil means "never asked". |
| + Alerts *AlertDelivery |
| // MemTotalBytes, MemAvailBytes, Load1, and DiskFreeBytes are the raw |
| // numbers MemFrac/DiskFrac above are already computed FROM — CollectBox |
| @@ -185,6 +188,13 @@ type BoxMetrics struct { |
| Platform string `json:"platform,omitempty"` // runtime.GOOS-runtime.GOARCH |
| SystemChannel string `json:"system_channel,omitempty"` // CARLOS_SYSTEM_CHANNEL |
| System *SystemState `json:"system,omitempty"` // last self-update outcome |
| + // Alerts is where this box's own alert rules (metrics.Evaluate) can send |
| + // a fire or clear, and how many sends have failed since the agent |
| + // started. nil means the box did not say (a snapshot from before the |
| + // field existed), which is not the same as an AlertDelivery whose sinks |
| + // are both off: that is a box that says it has nowhere to deliver. |
| + // Private, like the rest of this block: BuildFleet never copies Metrics. |
| + Alerts *AlertDelivery `json:"alerts,omitempty"` |
| // InstanceID is the cloud machine this box IS — read from the instance |
| // metadata service by the box itself (status.InstanceID). "" means no |
| // cloud machine owns it: a Pi, a laptop, someone else's hardware. |
| @@ -270,6 +280,30 @@ type AliasReading struct { |
| CertExpiresAt int64 `json:"cert_expires_at,omitempty"` |
| } |
| +// AlertDelivery is the box's own answer to "if a rule fires, does anyone |
| +// hear it": which sinks the agent has, and how many sends to them have |
| +// failed. It exists because an unset CARLOS_ALERT_WEBHOOK used to be |
| +// silent, and the flagship's disk, memory and cpu alerts went nowhere for |
| +// weeks (discussion #192). Webhook is CARLOS_ALERT_WEBHOOK set. SNS is |
| +// CARLOS_ALERT_SNS_TOPIC_ARN set to a topic ARN the agent can use; a |
| +// malformed value reads false, because it is not a sink. SNS is the state |
| +// of the sink the agent actually built, not of the env: when a topic ARN is |
| +// set but the agent could not build its SNS client (no region, a missing |
| +// profile), SNS is false and SNSUnavailable is true, so the box never reads |
| +// as having a sink it cannot use. Failures count sends (and that failed |
| +// build) since the agent process started, so a role without sns:Publish |
| +// shows up here and not only in the journal. |
| +type AlertDelivery struct { |
| + Webhook bool `json:"webhook"` |
| + SNS bool `json:"sns"` |
| + SNSUnavailable bool `json:"sns_unavailable,omitempty"` |
| + WebhookFailures int64 `json:"webhook_failures,omitempty"` |
| + SNSFailures int64 `json:"sns_failures,omitempty"` |
| +} |
| + |
| +// Delivered is whether at least one sink is configured. |
| +func (a AlertDelivery) Delivered() bool { return a.Webhook || a.SNS } |
| + |
| // RouteCounts is aggregate-only: no host, no socket path, just how many. |
| // Awake/Asleep stay null together whenever the socket table can't be read in |
| // full — partial knowledge published as a count would read as more certain |
| @@ -616,13 +650,14 @@ func CollectBox(routes []registry.Route, p Probes, now time.Time) Box { |
| p.DiskFreeBytes != nil || p.Release != "" || p.Tunnel != "" || |
| p.InstanceID != "" || p.Platform != "" || p.SystemChannel != "" || p.System != nil || |
| len(routeVersions) > 0 || len(aliasReadings) > 0 || len(p.Upstreams) > 0 || |
| - len(routeDBBytes) > 0 { |
| + len(routeDBBytes) > 0 || p.Alerts != nil { |
| box.Metrics = &BoxMetrics{ |
| MemTotalBytes: p.MemTotalBytes, MemAvailBytes: p.MemAvailBytes, |
| Load1: p.Load1, DiskFreeBytes: p.DiskFreeBytes, |
| Release: p.Release, Tunnel: p.Tunnel, |
| Platform: p.Platform, SystemChannel: p.SystemChannel, System: p.System, |
| InstanceID: p.InstanceID, |
| + Alerts: p.Alerts, |
| RouteVersions: routeVersions, |
| Aliases: aliasReadings, |
| // Carried through as the prober handed it over: proberSet.State |
diff --git a/internal/status/snapshot_test.go b/internal/status/snapshot_test.go| index e9518cdeb..5c11a933f 100644 |
| --- a/internal/status/snapshot_test.go |
| +++ b/internal/status/snapshot_test.go |
| @@ -48,3 +48,44 @@ func TestBoxMetricsCarryThePlatformSelfUpdateState(t *testing.T) { |
| t.Fatalf("fleet.json leaked the private metrics: %s", fb) |
| } |
| } |
| + |
| +// TestBoxMetricsCarryAlertDelivery pins discussion #192's visibility ask: a |
| +// box says whether its own alert rules have anywhere to deliver, an |
| +// all-off answer is published (it is the answer that matters), the field |
| +// alone makes a metrics block, and the public fleet feed never sees it. |
| +func TestBoxMetricsCarryAlertDelivery(t *testing.T) { |
| + at := time.Date(2026, 10, 5, 9, 0, 0, 0, time.UTC) |
| + |
| + none := CollectBox(nil, Probes{Alerts: &AlertDelivery{}}, at) |
| + if none.Metrics == nil || none.Metrics.Alerts == nil || none.Metrics.Alerts.Delivered() { |
| + t.Fatalf("an all-off answer must publish as not delivered: %+v", none.Metrics) |
| + } |
| + out, _ := json.Marshal(none) |
| + if !strings.Contains(string(out), `"alerts":{"webhook":false,"sns":false}`) { |
| + t.Errorf("json lacks the explicit all-off alerts block: %s", out) |
| + } |
| + |
| + unknown := CollectBox(nil, Probes{Release: "v1"}, at) |
| + if unknown.Metrics == nil || unknown.Metrics.Alerts != nil { |
| + t.Errorf("an unasked box must publish no alerts key: %+v", unknown.Metrics) |
| + } |
| + |
| + sns := CollectBox(nil, Probes{Alerts: &AlertDelivery{SNS: true, SNSFailures: 2}}, at) |
| + out, _ = json.Marshal(sns) |
| + if !strings.Contains(string(out), `"sns":true`) || !strings.Contains(string(out), `"sns_failures":2`) { |
| + t.Errorf("json lacks the sns sink or its failures: %s", out) |
| + } |
| + |
| + s := release.NewDirStore(t.TempDir()) |
| + if err := WriteBox(context.Background(), s, none); err != nil { |
| + t.Fatal(err) |
| + } |
| + f, err := BuildFleet(context.Background(), s, nil, at, 5*time.Minute) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + fb, _ := json.Marshal(f) |
| + if strings.Contains(string(fb), "alerts") || strings.Contains(string(fb), "webhook") { |
| + t.Fatalf("fleet.json leaked the alert-sink state: %s", fb) |
| + } |
| +} |
diff --git a/statuslive_test.go b/statuslive_test.go| index 028bae66d..93677c05e 100644 |
| --- a/statuslive_test.go |
| +++ b/statuslive_test.go |
| @@ -99,6 +99,51 @@ func TestPrintLiveBoxesSortsByLabel(t *testing.T) { |
| } |
| } |
| +// TestPrintLiveBoxesSaysWhetherAlertsAreDelivered is discussion #192's |
| +// visibility ask: a box with no alert sink must read as not delivered in the |
| +// operator's own listing, not as silence. A box that predates the field |
| +// prints a dash, since it did not say. A sink whose client could not be |
| +// built reads as unavailable, never as a working sink. |
| +func TestPrintLiveBoxesSaysWhetherAlertsAreDelivered(t *testing.T) { |
| + now := time.Now() |
| + var buf bytes.Buffer |
| + printLiveBoxes(&buf, []status.Box{ |
| + {GeneratedAt: now, Label: "a-none", Metrics: &status.BoxMetrics{Alerts: &status.AlertDelivery{}}}, |
| + {GeneratedAt: now, Label: "b-sns", Metrics: &status.BoxMetrics{Alerts: &status.AlertDelivery{SNS: true}}}, |
| + {GeneratedAt: now, Label: "c-both", Metrics: &status.BoxMetrics{Alerts: &status.AlertDelivery{Webhook: true, SNS: true, SNSFailures: 3}}}, |
| + {GeneratedAt: now, Label: "d-old", Metrics: &status.BoxMetrics{Tunnel: "connected"}}, |
| + {GeneratedAt: now, Label: "e-unavail", Metrics: &status.BoxMetrics{Alerts: &status.AlertDelivery{SNSUnavailable: true, SNSFailures: 1}}}, |
| + {GeneratedAt: now, Label: "f-hook-unavail", Metrics: &status.BoxMetrics{Alerts: &status.AlertDelivery{Webhook: true, SNSUnavailable: true}}}, |
| + }, now) |
| + want := map[string]string{ |
| + "a-none": "alerts=not-delivered", |
| + "b-sns": "alerts=sns\n", |
| + "c-both": "alerts=webhook+sns (sns failed 3)", |
| + "d-old": "alerts=-", |
| + "e-unavail": "alerts=sns-unavailable (sns failed 1)", |
| + "f-hook-unavail": "alerts=webhook (sns-unavailable)", |
| + } |
| + lines := strings.Split(strings.TrimRight(buf.String(), "\n"), "\n") |
| + if len(lines) != len(want) { |
| + t.Fatalf("got %d rows, want %d:\n%s", len(lines), len(want), buf.String()) |
| + } |
| + for label, frag := range want { |
| + var rows []string |
| + for _, line := range lines { |
| + if strings.HasPrefix(line, label+" ") { |
| + rows = append(rows, line) |
| + } |
| + } |
| + if len(rows) != 1 { |
| + t.Errorf("%s: appears on %d rows, want exactly 1:\n%s", label, len(rows), buf.String()) |
| + continue |
| + } |
| + if !strings.Contains(rows[0]+"\n", frag) { |
| + t.Errorf("%s: want %q in %q", label, frag, rows[0]) |
| + } |
| + } |
| +} |
| + |
| func TestHumanBytesAndFloat(t *testing.T) { |
| for _, c := range []struct { |
| in *int64 |