| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "math" |
| 9 | "strings" |
| 10 | "sync/atomic" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/alerting" |
| 14 | "github.com/carlosframework/platform/internal/metrics" |
| 15 | "github.com/carlosframework/platform/internal/release" |
| 16 | "github.com/carlosframework/platform/internal/status" |
| 17 | ) |
| 18 | |
| 19 | // Alert sinks: where a firing or clearing transition from metrics.Evaluate |
| 20 | // goes once alertTick has published the new state to the bucket. |
| 21 | // |
| 22 | // Two sinks, either, both, or neither: |
| 23 | // |
| 24 | // CARLOS_ALERT_WEBHOOK the event as JSON, POSTed (httpPostWebhook) |
| 25 | // CARLOS_ALERT_SNS_TOPIC_ARN a plain-text subject and message, published |
| 26 | // to that SNS topic with the box's instance role |
| 27 | // |
| 28 | // SNS is here because a webhook cannot reach it: Publish needs a SigV4 |
| 29 | // signature, which only the AWS SDK on a box holding the role can make. The |
| 30 | // flagship's topics (carlos-flagship-<region>-alerts) already carry confirmed |
| 31 | // email subscriptions, so this is the path from a disk at 91% to an inbox |
| 32 | // (discussion #192). |
| 33 | // |
| 34 | // Delivery is best effort and never fatal. A sink that fails is logged and |
| 35 | // counted, the other sink still runs, and the tick carries on. Failed sends |
| 36 | // are not rescheduled by a later tick: Evaluate emits an event only on a |
| 37 | // transition, so the next tick has no event to resend. (The SDK itself tries a |
| 38 | // retryable failure up to three times inside one publish, and a failed bucket |
| 39 | // write in alertTick repeats the whole transition next tick, delivery |
| 40 | // included.) The count is what makes a lost send visible, through |
| 41 | // status.AlertDelivery in the box snapshot. |
| 42 | // |
| 43 | // The delivery itself (the webhook post, the SNS publish and its timeout) |
| 44 | // is internal/alerting's, shared with the console; this file formats the |
| 45 | // box's alerts and counts their failures. The same sinks carry the |
| 46 | // operator pages of the moves code (PageOperator, moves spec §13). |
| 47 | |
| 48 | // snsPublisher is the one SNS call the sinks make. |
| 49 | type snsPublisher = alerting.Publisher |
| 50 | |
| 51 | // alertCounters counts failed sends since the process started, and holds the |
| 52 | // one fact the status loop cannot derive from the env: whether the SNS client |
| 53 | // was actually built. They are separate goroutines, so this is their seam. |
| 54 | type alertCounters struct { |
| 55 | webhook atomic.Int64 |
| 56 | sns atomic.Int64 |
| 57 | // snsInit is snsInitUnknown until buildAlertSinks has run for a valid |
| 58 | // topic ARN, then snsInitOK or snsInitFailed. |
| 59 | snsInit atomic.Int32 |
| 60 | } |
| 61 | |
| 62 | const ( |
| 63 | snsInitUnknown int32 = iota |
| 64 | snsInitOK |
| 65 | snsInitFailed |
| 66 | ) |
| 67 | |
| 68 | // processAlertCounters is the process-wide count the metrics loop writes |
| 69 | // and the status loop reads. They are separate goroutines with no other |
| 70 | // shared state, so this is the seam between them. |
| 71 | var processAlertCounters alertCounters |
| 72 | |
| 73 | // alertSinks is the configured delivery for one agent: alerting's sinks, |
| 74 | // counting their failures in counts. |
| 75 | type alertSinks struct { |
| 76 | sinks alerting.Sinks |
| 77 | counts *alertCounters |
| 78 | } |
| 79 | |
| 80 | // newAlertSinks builds a sink set. An empty webhookURL or topicARN switches |
| 81 | // that sink off; a nil publisher does the same for SNS. |
| 82 | func newAlertSinks(webhookURL string, post func(url string, body []byte) error, |
| 83 | topicARN string, pub snsPublisher, counts *alertCounters) *alertSinks { |
| 84 | if counts == nil { |
| 85 | counts = &processAlertCounters |
| 86 | } |
| 87 | return &alertSinks{counts: counts, sinks: alerting.Sinks{WebhookURL: webhookURL, Post: post, TopicARN: topicARN, Pub: pub, |
| 88 | WebhookFailed: func() { counts.webhook.Add(1) }, SNSFailed: func() { counts.sns.Add(1) }}} |
| 89 | } |
| 90 | |
| 91 | // deliver sends one transition to every configured sink. Safe on a nil |
| 92 | // receiver, which means "no sinks". |
| 93 | func (a *alertSinks) deliver(ctx context.Context, rules metrics.Rules, at time.Time, ev metrics.Event) { |
| 94 | if a == nil { |
| 95 | return |
| 96 | } |
| 97 | if a.sinks.HasWebhook() { |
| 98 | evBody, err := json.Marshal(ev) |
| 99 | if err != nil { |
| 100 | log.Printf("host: alert event %s: marshal: %v", ev.Rule, err) |
| 101 | } else if err := a.sinks.PostWebhook(evBody); err != nil { |
| 102 | log.Printf("host: alert webhook post (rule %s): %v", ev.Rule, err) |
| 103 | } |
| 104 | } |
| 105 | if a.sinks.HasSNS() { |
| 106 | subject, message := formatAlert(rules, at, ev) |
| 107 | if err := a.sinks.PublishSNS(ctx, subject, message); err != nil { |
| 108 | log.Printf("host: alert sns publish (rule %s): %v", ev.Rule, err) |
| 109 | } |
| 110 | } |
| 111 | } |
| 112 | |
| 113 | // PageOperator pages an operator through this agent's sinks: the same |
| 114 | // webhook and SNS topic deliver uses, its failures counted with theirs |
| 115 | // (alerting.Sinks.Page). Safe on a nil receiver, which means "no sinks": |
| 116 | // nil, and one line in the journal. |
| 117 | func (a *alertSinks) PageOperator(ctx context.Context, level, subject, body string) error { |
| 118 | if a == nil { |
| 119 | return (*alerting.Sinks)(nil).Page(ctx, level, subject, body) |
| 120 | } |
| 121 | return a.sinks.Page(ctx, level, subject, body) |
| 122 | } |
| 123 | |
| 124 | // PageOperator pages an operator on the sinks the metrics loop armed |
| 125 | // (armPaging): the moves code's paging primitive (moves spec §13). Before |
| 126 | // they are armed, or with none configured, it returns nil and says so once. |
| 127 | func PageOperator(ctx context.Context, level, subject, body string) error { |
| 128 | return alerting.Page(ctx, level, subject, body) |
| 129 | } |
| 130 | |
| 131 | // armAgentAlerts builds the agent's alert sinks from the env and arms |
| 132 | // paging on them (armPaging). runAgent calls it once at startup, before |
| 133 | // anything that can run a moves pass, and hands the same instance to the |
| 134 | // metrics loop, so arming never waits on the metrics store. |
| 135 | func armAgentAlerts(ctx context.Context, getenv func(string) string, counts *alertCounters, newClient snsClientFactory) *alertSinks { |
| 136 | sinks := buildAlertSinks(ctx, getenv, counts, newClient) |
| 137 | armPaging(sinks) |
| 138 | return sinks |
| 139 | } |
| 140 | |
| 141 | // armPaging makes a the sinks PageOperator pages on. nil means none. |
| 142 | func armPaging(a *alertSinks) { |
| 143 | if a == nil { |
| 144 | alerting.Arm(nil) |
| 145 | return |
| 146 | } |
| 147 | alerting.Arm(&a.sinks) |
| 148 | } |
| 149 | |
| 150 | // deliverRelease sends a D1 refusal to every configured sink. The webhook |
| 151 | // body is the record with kind "release_refused"; SNS gets a sentence. |
| 152 | func (a *alertSinks) deliverRelease(ctx context.Context, r release.Refusal) { |
| 153 | if a == nil { |
| 154 | return |
| 155 | } |
| 156 | if a.sinks.HasWebhook() { |
| 157 | body, err := json.Marshal(struct { |
| 158 | Kind string `json:"kind"` |
| 159 | // The record's own Kind is shadowed by the envelope's "kind" |
| 160 | // above, so it travels as refusal_kind. |
| 161 | RefusalKind string `json:"refusal_kind,omitempty"` |
| 162 | release.Refusal |
| 163 | }{"release_refused", r.Kind, r}) |
| 164 | if err == nil { |
| 165 | if err := a.sinks.PostWebhook(body); err != nil { |
| 166 | log.Printf("host: alert webhook post (release refusal %s): %v", r.Host, err) |
| 167 | } |
| 168 | } |
| 169 | } |
| 170 | if a.sinks.HasSNS() { |
| 171 | subject, msg := formatReleaseAlert(r) |
| 172 | if err := a.sinks.PublishSNS(ctx, subject, msg); err != nil { |
| 173 | log.Printf("host: alert sns publish (release refusal %s): %v", r.Host, err) |
| 174 | } |
| 175 | } |
| 176 | } |
| 177 | |
| 178 | // formatReleaseAlert is a refusal's SNS subject and message. A recovery that |
| 179 | // failed reads as one in the subject (spec §2.6): "rolled back" over a box |
| 180 | // whose restored build never came up would send the operator back to bed. |
| 181 | func formatReleaseAlert(r release.Refusal) (subject, msg string) { |
| 182 | if r.Kind == release.RefusalWrongProgram { |
| 183 | // Nothing started and nothing was rolled back: the box never left its |
| 184 | // old build, and "--retry" would be refused the same way. A version's |
| 185 | // manifest is immutable, so the way out is a NEW version. |
| 186 | subject = fmt.Sprintf("%s: %s refused, wrong program (still on %s)", r.Host, r.Version, r.RestoredVersion) |
| 187 | if len(subject) > 100 { // SNS subject limit |
| 188 | subject = subject[:100] |
| 189 | } |
| 190 | msg = fmt.Sprintf("%s refused %s at %s: %s\n\nShip a new version with carlos ship --program-change and promote it, if the change is intended, or promote a different build.", |
| 191 | r.Host, r.Version, r.RefusedAt.Format(time.RFC3339), r.Reason) |
| 192 | return subject, msg |
| 193 | } |
| 194 | if r.Kind == release.RefusalNotProven { |
| 195 | // D3's binding check: the same shape as the wrong program. Nothing |
| 196 | // started and nothing was rolled back, and "--retry" would re-sign |
| 197 | // the pointer over the same staged bytes. |
| 198 | still := r.RestoredVersion |
| 199 | if still == "" { |
| 200 | still = "nothing" |
| 201 | } |
| 202 | subject = fmt.Sprintf("%s: %s refused, staged bytes do not match dev's proof (still on %s)", r.Host, r.Version, still) |
| 203 | if len(subject) > 100 { // SNS subject limit |
| 204 | subject = subject[:100] |
| 205 | } |
| 206 | msg = fmt.Sprintf("%s refused %s at %s: %s\n\nThe staged bytes do not match dev's proof, so %s was never pointed live. The box is still on %s. Roll the exact bytes dev proved.", |
| 207 | r.Host, r.Version, r.RefusedAt.Format(time.RFC3339), strings.TrimPrefix(r.Reason, "not proven: "), r.Version, still) |
| 208 | return subject, msg |
| 209 | } |
| 210 | if r.Recovered { |
| 211 | subject = fmt.Sprintf("%s: %s did not start, rolled back to %s", r.Host, r.Version, r.RestoredVersion) |
| 212 | } else { |
| 213 | subject = fmt.Sprintf("%s: %s did not start and the rollback did not come up either", r.Host, r.Version) |
| 214 | } |
| 215 | if len(subject) > 100 { // SNS subject limit |
| 216 | subject = subject[:100] |
| 217 | } |
| 218 | msg = fmt.Sprintf("%s refused %s at %s: %s", r.Host, r.Version, r.RefusedAt.Format(time.RFC3339), r.Reason) |
| 219 | if !r.Recovered { |
| 220 | msg += "\n\nThe rollback did not come up either: " + recoveryError(r) |
| 221 | } |
| 222 | msg += fmt.Sprintf("\n\nRetry with: carlos promote --retry %s <channel>", r.Version) |
| 223 | return subject, msg |
| 224 | } |
| 225 | |
| 226 | // recoveryError is why a refusal's recovery failed, never empty for one that |
| 227 | // did: a record written before the outcome was recorded has none to give. |
| 228 | func recoveryError(r release.Refusal) string { |
| 229 | if r.RecoveryError != "" { |
| 230 | return r.RecoveryError |
| 231 | } |
| 232 | return "outcome not recorded" |
| 233 | } |
| 234 | |
| 235 | // parseAlertTopicARN checks CARLOS_ALERT_SNS_TOPIC_ARN and returns the |
| 236 | // region the topic lives in (alerting.ParseTopicARN). |
| 237 | func parseAlertTopicARN(s string) (region string, err error) { return alerting.ParseTopicARN(s) } |
| 238 | |
| 239 | // alertDeliveryProbe is the box's answer to "do my alert rules have anywhere |
| 240 | // to go". The webhook and the ARN's shape come from the same env |
| 241 | // buildAlertSinks reads; whether the SNS client was actually built comes from |
| 242 | // counts, which buildAlertSinks fills in. So the snapshot reports the sink the |
| 243 | // agent really holds: a topic ARN that does not parse is not a sink, and one |
| 244 | // whose client failed to build reads SNS=false, SNSUnavailable=true. |
| 245 | func alertDeliveryProbe(getenv func(string) string, counts *alertCounters) *status.AlertDelivery { |
| 246 | if counts == nil { |
| 247 | counts = &processAlertCounters |
| 248 | } |
| 249 | out := &status.AlertDelivery{ |
| 250 | Webhook: strings.TrimSpace(getenv("CARLOS_ALERT_WEBHOOK")) != "", |
| 251 | WebhookFailures: counts.webhook.Load(), |
| 252 | SNSFailures: counts.sns.Load(), |
| 253 | } |
| 254 | if _, err := parseAlertTopicARN(getenv("CARLOS_ALERT_SNS_TOPIC_ARN")); err == nil { |
| 255 | // Only an initialised client is a sink. snsInitUnknown (not built |
| 256 | // yet, or the metrics loop never got there) is neither available |
| 257 | // nor unavailable: the box has no working SNS sink to report. |
| 258 | switch counts.snsInit.Load() { |
| 259 | case snsInitOK: |
| 260 | out.SNS = true |
| 261 | case snsInitFailed: |
| 262 | out.SNSUnavailable = true |
| 263 | } |
| 264 | } |
| 265 | return out |
| 266 | } |
| 267 | |
| 268 | // snsClientFactory builds the publisher for a region. It is a parameter so |
| 269 | // tests can make client construction fail without an AWS config. |
| 270 | type snsClientFactory func(ctx context.Context, region string) (snsPublisher, error) |
| 271 | |
| 272 | // newSNSClient is the production factory: the default credential chain (the |
| 273 | // box's instance role), aimed at the topic's own region. |
| 274 | func newSNSClient(ctx context.Context, region string) (snsPublisher, error) { |
| 275 | return alerting.NewSNSClient(ctx, region) |
| 276 | } |
| 277 | |
| 278 | // buildAlertSinks reads the sink env and builds the sinks. A bad topic ARN |
| 279 | // is logged once and leaves SNS off; a client that cannot be built does the |
| 280 | // same, and is also recorded in counts (state and one counted failure) so the |
| 281 | // status snapshot reports the sink the agent really has. Neither stops the |
| 282 | // metrics loop. With nothing configured it says so in the journal, because |
| 283 | // that is the state that used to be silent. |
| 284 | func buildAlertSinks(ctx context.Context, getenv func(string) string, counts *alertCounters, newClient snsClientFactory) *alertSinks { |
| 285 | webhook := strings.TrimSpace(getenv("CARLOS_ALERT_WEBHOOK")) |
| 286 | topic := strings.TrimSpace(getenv("CARLOS_ALERT_SNS_TOPIC_ARN")) |
| 287 | |
| 288 | var pub snsPublisher |
| 289 | if topic != "" { |
| 290 | region, err := parseAlertTopicARN(topic) |
| 291 | if err != nil { |
| 292 | log.Printf("agent: alerts: CARLOS_ALERT_SNS_TOPIC_ARN is not usable, SNS sink off: %v", err) |
| 293 | topic = "" |
| 294 | } else if p, err := newClient(ctx, region); err != nil { |
| 295 | log.Printf("agent: alerts: build SNS client, SNS sink off: %v", err) |
| 296 | counts.sns.Add(1) |
| 297 | counts.snsInit.Store(snsInitFailed) |
| 298 | topic = "" |
| 299 | } else { |
| 300 | counts.snsInit.Store(snsInitOK) |
| 301 | pub = p |
| 302 | } |
| 303 | } |
| 304 | |
| 305 | switch { |
| 306 | case webhook == "" && topic == "": |
| 307 | 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") |
| 308 | case webhook != "" && topic != "": |
| 309 | log.Printf("agent: alerts: delivering to webhook and SNS topic %s", topic) |
| 310 | case topic != "": |
| 311 | log.Printf("agent: alerts: delivering to SNS topic %s", topic) |
| 312 | default: |
| 313 | log.Printf("agent: alerts: delivering to webhook") |
| 314 | } |
| 315 | return newAlertSinks(webhook, httpPostWebhook, topic, pub, counts) |
| 316 | } |
| 317 | |
| 318 | // formatAlert renders one transition as an email: a one-line subject and a |
| 319 | // short plain-text body. The inputs are the event, the rule levels and a |
| 320 | // time, and nothing else, so no URL, ARN or credential can reach the |
| 321 | // message. The subject is ASCII with no newline and under 100 characters, |
| 322 | // which is what SNS accepts. |
| 323 | func formatAlert(rules metrics.Rules, at time.Time, ev metrics.Event) (subject, body string) { |
| 324 | subjectLine := alertSubject(ev) |
| 325 | var b strings.Builder |
| 326 | if ev.Firing { |
| 327 | fmt.Fprintf(&b, "%s.\n\n", alertSentence(ev)) |
| 328 | } else { |
| 329 | fmt.Fprintf(&b, "%s.\n\n", alertClearSentence(ev)) |
| 330 | } |
| 331 | if level, ok := alertLevel(rules, ev.Rule); ok { |
| 332 | fmt.Fprintf(&b, "the alert level is %s.\n", level) |
| 333 | } |
| 334 | if ev.Firing { |
| 335 | if d := alertSustain(rules, ev.Rule); d > 0 { |
| 336 | fmt.Fprintf(&b, "it stayed above that level for %d minutes before this alert was raised.\n", int(d.Minutes())) |
| 337 | } |
| 338 | } |
| 339 | fmt.Fprintf(&b, "this was reported at %s.\n", at.UTC().Format("2006-01-02 15:04 MST")) |
| 340 | if ev.Firing { |
| 341 | b.WriteString("you will get another email when it clears.\n") |
| 342 | } |
| 343 | fmt.Fprintf(&b, "\nrule: %s\nbox: %s\n", ev.Rule, ev.Box) |
| 344 | return subjectLine, b.String() |
| 345 | } |
| 346 | |
| 347 | func alertSubject(ev metrics.Event) string { |
| 348 | if ev.Firing { |
| 349 | return "carlos alert: " + alertSentence(ev) |
| 350 | } |
| 351 | return "carlos alert cleared: " + alertClearSentence(ev) |
| 352 | } |
| 353 | |
| 354 | // alertSentence is the lower-case "what is wrong" clause, shared by the |
| 355 | // subject and the body. |
| 356 | func alertSentence(ev metrics.Event) string { |
| 357 | switch ev.Rule { |
| 358 | case "disk": |
| 359 | return fmt.Sprintf("disk is %s full on %s", pct(ev.Value), ev.Box) |
| 360 | case "mem": |
| 361 | return fmt.Sprintf("memory use is %s on %s", pct(ev.Value), ev.Box) |
| 362 | case "cpu": |
| 363 | return fmt.Sprintf("cpu use is %s on %s", pct(ev.Value), ev.Box) |
| 364 | case "lag": |
| 365 | return fmt.Sprintf("mail relay lag is %d seconds on %s", int64(ev.Value), ev.Box) |
| 366 | case "geoip": |
| 367 | return geoipAlertSentence(ev) |
| 368 | default: |
| 369 | return fmt.Sprintf("%s is firing on %s", ev.Rule, ev.Box) |
| 370 | } |
| 371 | } |
| 372 | |
| 373 | func alertClearSentence(ev metrics.Event) string { |
| 374 | switch ev.Rule { |
| 375 | case "disk": |
| 376 | return fmt.Sprintf("disk is back to %s full on %s", pct(ev.Value), ev.Box) |
| 377 | case "mem": |
| 378 | return fmt.Sprintf("memory use is back to %s on %s", pct(ev.Value), ev.Box) |
| 379 | case "cpu": |
| 380 | return fmt.Sprintf("cpu use is back to %s on %s", pct(ev.Value), ev.Box) |
| 381 | case "lag": |
| 382 | return fmt.Sprintf("mail relay lag is back to %d seconds on %s", int64(ev.Value), ev.Box) |
| 383 | case "geoip": |
| 384 | return fmt.Sprintf("geoip on %s is back to normal", ev.Box) |
| 385 | default: |
| 386 | return fmt.Sprintf("%s has cleared on %s", ev.Rule, ev.Box) |
| 387 | } |
| 388 | } |
| 389 | |
| 390 | func alertLevel(r metrics.Rules, rule string) (string, bool) { |
| 391 | switch rule { |
| 392 | case "disk": |
| 393 | return pct(r.DiskFrac), true |
| 394 | case "mem": |
| 395 | return pct(r.MemFrac), true |
| 396 | case "cpu": |
| 397 | return pct(r.CPUFrac), true |
| 398 | case "lag": |
| 399 | return fmt.Sprintf("%d seconds", r.LagSecs), true |
| 400 | } |
| 401 | return "", false |
| 402 | } |
| 403 | |
| 404 | func alertSustain(r metrics.Rules, rule string) time.Duration { |
| 405 | switch rule { |
| 406 | case "mem": |
| 407 | return r.MemSustain |
| 408 | case "cpu": |
| 409 | return r.CPUSustain |
| 410 | } |
| 411 | return 0 |
| 412 | } |
| 413 | |
| 414 | // geoipAlertSentence reads the geoip rule's reason code (geoip spec §10.4). |
| 415 | func geoipAlertSentence(ev metrics.Event) string { |
| 416 | switch int(ev.Value) { |
| 417 | case 1: |
| 418 | return fmt.Sprintf("the geoip database on %s is more than 45 days old", ev.Box) |
| 419 | case 2: |
| 420 | return fmt.Sprintf("the GeoLite2 database on %s expires within 9 days", ev.Box) |
| 421 | case 3: |
| 422 | return fmt.Sprintf("the geoip database on %s has expired and was removed", ev.Box) |
| 423 | case 4: |
| 424 | return fmt.Sprintf("%s has not loaded the published geoip database for over an hour", ev.Box) |
| 425 | case 5: |
| 426 | return fmt.Sprintf("the geoip feed has failed 7 times in a row (seen from %s)", ev.Box) |
| 427 | } |
| 428 | return fmt.Sprintf("geoip needs attention on %s", ev.Box) |
| 429 | } |
| 430 | |
| 431 | // pct renders a 0..1 fraction as a whole percent. |
| 432 | func pct(f float64) string { return fmt.Sprintf("%d%%", int(math.Round(f*100))) } |
| 433 | |
amadan