| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "sort" |
| 9 | "strconv" |
| 10 | "strings" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/metrics" |
| 14 | ) |
| 15 | |
| 16 | // The keymail half of alert delivery lives here rather than on the box, |
| 17 | // because the only keymail send path that exists is the console's per-user |
| 18 | // OAuth grant (Keymail.SendApp) and the agent has no keymail identity. The |
| 19 | // box judges itself and publishes alerts/boxes/<box>.json; this watcher |
| 20 | // polls that prefix, diffs each rule's firing state against what it last |
| 21 | // told each operator (the alert_seen table), and mails the difference — |
| 22 | // each operator with that operator's OWN grant token, so nobody is sent |
| 23 | // mail on a credential they did not personally issue. |
| 24 | // |
| 25 | // The diff is per RECIPIENT, not per rule, because delivery fails per |
| 26 | // recipient: a success records that operator's row and a failure leaves |
| 27 | // only theirs unrecorded, so the next pass retries exactly the operator |
| 28 | // who missed out. One operator's revoked grant can no longer re-page every |
| 29 | // healthy operator on the fleet, tick after tick. |
| 30 | // |
| 31 | // The failure posture throughout is "never go silent": an unreadable or |
| 32 | // unparsable alert object is logged and skipped, an undelivered transition |
| 33 | // stays pending for that operator until it lands, and no path here can |
| 34 | // panic a console that is otherwise serving fine. |
| 35 | |
| 36 | // alertWatchInterval is the poll cadence from the spec: a minute is fast |
| 37 | // enough for a disk filling up and slow enough to be free. |
| 38 | const alertWatchInterval = time.Minute |
| 39 | |
| 40 | // alertPrefix is the bucket prefix the agent publishes AlertState under |
| 41 | // (metrics.AlertKey). |
| 42 | const alertPrefix = "alerts/boxes/" |
| 43 | |
| 44 | // StartAlertWatcher runs alertPass immediately and then every minute until |
| 45 | // ctx is done. Started from cmd/console's serve(), deliberately NOT from |
| 46 | // NewServer: tests construct servers by the dozen and none of them should |
| 47 | // grow a background goroutine or a keymail send. |
| 48 | func (s *Server) StartAlertWatcher(ctx context.Context) { |
| 49 | t := time.NewTicker(alertWatchInterval) |
| 50 | defer t.Stop() |
| 51 | for { |
| 52 | if err := s.alertPass(ctx); err != nil { |
| 53 | log.Printf("console: alert watch: %v", err) |
| 54 | } |
| 55 | select { |
| 56 | case <-ctx.Done(): |
| 57 | return |
| 58 | case <-t.C: |
| 59 | } |
| 60 | } |
| 61 | } |
| 62 | |
| 63 | // operatorGrant is one recipient and the token the console sends AS them |
| 64 | // with. The token is never logged. |
| 65 | type operatorGrant struct { |
| 66 | address string |
| 67 | token string |
| 68 | } |
| 69 | |
| 70 | // alertRun is one pass's working state: who to mail, and the operators with |
| 71 | // no grant to mail through — reported at most once per pass, and only if |
| 72 | // the pass had something to say, so a fleet at rest logs nothing. |
| 73 | type alertRun struct { |
| 74 | to []operatorGrant |
| 75 | missing []string |
| 76 | noted bool |
| 77 | } |
| 78 | |
| 79 | // noteMissing names the operators the console cannot write as, at most once |
| 80 | // per pass, called only when the pass actually has something to deliver. |
| 81 | func (r *alertRun) noteMissing() { |
| 82 | if r.noted || len(r.missing) == 0 { |
| 83 | return |
| 84 | } |
| 85 | r.noted = true |
| 86 | log.Printf("console: alert watch: no keymail grant for %s — not notified (they must sign in to the console once)", |
| 87 | strings.Join(r.missing, ", ")) |
| 88 | } |
| 89 | |
| 90 | // noMailboxAddress is the synthetic alert_seen recipient standing for "no |
| 91 | // operator was mailable when this transition happened". Recording it stops |
| 92 | // the transition replaying — and being re-logged — every minute forever, |
| 93 | // at the honest cost that the alert is simply lost: an operator who gains a |
| 94 | // grant later is not caught up on it (see alertSeenFor). |
| 95 | const noMailboxAddress = "" |
| 96 | |
| 97 | // alertPass reads every alerts/boxes/*.json and, for each rule, diffs its |
| 98 | // firing state against what alert_seen says each operator was last told, |
| 99 | // mailing and recording per recipient. It returns an error only for a failure |
| 100 | // that makes the whole pass meaningless (listing the bucket, or reading the |
| 101 | // operator list); per-box and per-rule trouble is logged and stepped over. |
| 102 | func (s *Server) alertPass(ctx context.Context) error { |
| 103 | if s.cfg.Releases == nil || s.sendApp == nil { |
| 104 | return nil |
| 105 | } |
| 106 | keys, err := s.cfg.Releases.List(ctx, alertPrefix) |
| 107 | if err != nil { |
| 108 | return fmt.Errorf("list %s: %w", alertPrefix, err) |
| 109 | } |
| 110 | if len(keys) == 0 { |
| 111 | return nil // no box has ever published an alert: nothing to do |
| 112 | } |
| 113 | run, err := s.alertRecipients() |
| 114 | if err != nil { |
| 115 | return err |
| 116 | } |
| 117 | for _, key := range keys { |
| 118 | if !strings.HasSuffix(key, ".json") { |
| 119 | continue |
| 120 | } |
| 121 | body, _, err := s.cfg.Releases.Get(ctx, key) |
| 122 | if err != nil { |
| 123 | log.Printf("console: alert watch: read %s: %v", key, err) |
| 124 | continue |
| 125 | } |
| 126 | var st metrics.AlertState |
| 127 | if err := json.Unmarshal(body, &st); err != nil { |
| 128 | log.Printf("console: alert watch: unparsable %s: %v", key, err) |
| 129 | continue |
| 130 | } |
| 131 | box := st.Box |
| 132 | if box == "" { |
| 133 | // Trust the key over an empty body field: the key is what the |
| 134 | // publisher's own AlertKey produced. |
| 135 | box = strings.TrimSuffix(strings.TrimPrefix(key, alertPrefix), ".json") |
| 136 | } |
| 137 | names := make([]string, 0, len(st.Rules)) |
| 138 | for name := range st.Rules { |
| 139 | names = append(names, name) |
| 140 | } |
| 141 | sort.Strings(names) |
| 142 | for _, name := range names { |
| 143 | s.alertRule(ctx, run, box, name, st.Rules[name], st.UpdatedAt) |
| 144 | } |
| 145 | } |
| 146 | return nil |
| 147 | } |
| 148 | |
| 149 | // alertRecipients resolves the operator list into recipients, splitting off |
| 150 | // the operators the console cannot write as. |
| 151 | func (s *Server) alertRecipients() (*alertRun, error) { |
| 152 | addrs, err := s.store.Operators() |
| 153 | if err != nil { |
| 154 | return nil, fmt.Errorf("operators: %w", err) |
| 155 | } |
| 156 | return s.grantRecipients(addrs) |
| 157 | } |
| 158 | |
| 159 | // grantRecipients resolves canonical addresses to their own Keymail grants. |
| 160 | // Alert delivery and one-shot console notifications share this lookup so both |
| 161 | // keep the same missing-grant and per-recipient-token behaviour. |
| 162 | func (s *Server) grantRecipients(addrs []string) (*alertRun, error) { |
| 163 | run := &alertRun{} |
| 164 | seen := make(map[string]bool) |
| 165 | for _, addr := range addrs { |
| 166 | addr = strings.ToLower(strings.TrimSpace(addr)) |
| 167 | if addr == "" || seen[addr] { |
| 168 | continue |
| 169 | } |
| 170 | seen[addr] = true |
| 171 | tok, err := s.store.GrantToken(addr) |
| 172 | if err != nil { |
| 173 | return nil, fmt.Errorf("grant token for %s: %w", addr, err) |
| 174 | } |
| 175 | if tok == "" { |
| 176 | run.missing = append(run.missing, addr) |
| 177 | continue |
| 178 | } |
| 179 | run.to = append(run.to, operatorGrant{address: addr, token: tok}) |
| 180 | } |
| 181 | return run, nil |
| 182 | } |
| 183 | |
| 184 | // alertRule handles one rule of one box: every mailable operator whose |
| 185 | // recorded state differs from the published one gets a keymail, and each |
| 186 | // send is recorded on its own. |
| 187 | func (s *Server) alertRule(ctx context.Context, run *alertRun, box, rule string, rs metrics.RuleState, updated time.Time) { |
| 188 | if len(run.to) == 0 { |
| 189 | // Nobody the console can write as. Record the transition against the |
| 190 | // synthetic recipient so it stops replaying, and say so out loud — |
| 191 | // this is the one path where an alert is genuinely dropped. |
| 192 | if s.alertDueFor(box, rule, noMailboxAddress, rs) { |
| 193 | run.noteMissing() |
| 194 | log.Printf("console: alert watch: %s/%s transition dropped — no operator holds a keymail grant", box, rule) |
| 195 | s.recordAlertSeenFor(box, rule, noMailboxAddress, rs) |
| 196 | } |
| 197 | return |
| 198 | } |
| 199 | var due []operatorGrant |
| 200 | for _, op := range run.to { |
| 201 | if s.alertDueFor(box, rule, op.address, rs) { |
| 202 | due = append(due, op) |
| 203 | } |
| 204 | } |
| 205 | if len(due) == 0 { |
| 206 | return |
| 207 | } |
| 208 | run.noteMissing() |
| 209 | subject, body := alertMessage(box, rule, rs, updated) |
| 210 | for _, op := range due { |
| 211 | if _, err := s.sendApp(ctx, op.token, op.address, subject, body); err != nil { |
| 212 | // The address is safe to log; the token never is. Leaving this |
| 213 | // operator's row unrecorded is the retry: the next pass mails |
| 214 | // them and nobody else. |
| 215 | log.Printf("console: alert watch: keymail %s/%s to %s: %v", box, rule, op.address, err) |
| 216 | continue |
| 217 | } |
| 218 | s.recordAlertSeenFor(box, rule, op.address, rs) |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | // geoipReasonRule is the dedup key for one geoip reason (geoip spec |
| 223 | // §10.4). The geoip rule's VALUE is a reason code, and a reason that |
| 224 | // changes while firing (expiring becomes expired) is news the operator |
| 225 | // must get, though firing never went false (code review R4-3). The plain |
| 226 | // "geoip" row still carries firing/cleared; one extra row per reason marks |
| 227 | // which reasons this address has been told about. |
| 228 | func geoipReasonRule(code float64) string { return fmt.Sprintf("geoip@%d", int(code)) } |
| 229 | |
| 230 | // alertDueFor is alertDue plus the geoip reason rule. |
| 231 | func (s *Server) alertDueFor(box, rule, address string, rs metrics.RuleState) bool { |
| 232 | due := s.alertDue(box, rule, address, rs.Firing) |
| 233 | if rule == "geoip" && rs.Firing && !due { |
| 234 | due = s.alertDue(box, geoipReasonRule(rs.Value), address, true) |
| 235 | } |
| 236 | return due |
| 237 | } |
| 238 | |
| 239 | // recordAlertSeenFor records the transition and, for geoip, which reason |
| 240 | // this address was LAST told about: exactly one reason row is true while |
| 241 | // firing and none after a clear, so a reason seen before (2, 3, back to 2, |
| 242 | // or a later episode) is news again (code review R5 on R4-3). |
| 243 | func (s *Server) recordAlertSeenFor(box, rule, address string, rs metrics.RuleState) { |
| 244 | s.recordAlertSeen(box, rule, address, rs.Firing) |
| 245 | if rule != "geoip" { |
| 246 | return |
| 247 | } |
| 248 | for code := 1; code <= 5; code++ { |
| 249 | s.recordAlertSeen(box, geoipReasonRule(float64(code)), address, rs.Firing && int(rs.Value) == code) |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | // alertDue reports whether address must be told that (box, rule) is now in |
| 254 | // this firing state, recording a quiet first sighting as it goes — nobody |
| 255 | // wants mail saying the disk is fine. |
| 256 | // |
| 257 | // An address the console has never written to about this rule falls back to |
| 258 | // the synthetic no-mailbox row: if the transition was already dropped for |
| 259 | // want of any grant at all, an operator who gains a grant afterwards is not |
| 260 | // retroactively paged about it. |
| 261 | func (s *Server) alertDue(box, rule, address string, firing bool) bool { |
| 262 | seen, known, err := s.store.AlertSeen(box, rule, address) |
| 263 | if err != nil { |
| 264 | log.Printf("console: alert watch: alert_seen %s/%s for %s: %v", box, rule, address, err) |
| 265 | return false |
| 266 | } |
| 267 | if !known && address != noMailboxAddress { |
| 268 | seen, known, err = s.store.AlertSeen(box, rule, noMailboxAddress) |
| 269 | if err != nil { |
| 270 | log.Printf("console: alert watch: alert_seen %s/%s: %v", box, rule, err) |
| 271 | return false |
| 272 | } |
| 273 | } |
| 274 | if known && seen == firing { |
| 275 | return false |
| 276 | } |
| 277 | if !known && !firing { |
| 278 | // First sighting of a quiet rule. Record it so a later fire is a |
| 279 | // transition; say nothing. |
| 280 | s.recordAlertSeen(box, rule, address, false) |
| 281 | return false |
| 282 | } |
| 283 | return true |
| 284 | } |
| 285 | |
| 286 | func (s *Server) recordAlertSeen(box, rule, address string, firing bool) { |
| 287 | if err := s.store.SetAlertSeen(box, rule, address, firing); err != nil { |
| 288 | log.Printf("console: alert watch: record %s/%s for %s: %v", box, rule, address, err) |
| 289 | } |
| 290 | } |
| 291 | |
| 292 | // alertValueText renders one rule's transition value in that rule's own |
| 293 | // unit. Every rule internal/metrics evaluates carries a fraction EXCEPT lag, |
| 294 | // which is keymail relay seconds (see metrics.Rules.LagSecs) — so a single |
| 295 | // percent formatter for all of them turns the first lag alert to fire into |
| 296 | // "92000%". Used by both operator-facing surfaces: the dashboard's alerts |
| 297 | // table and the alert mail below. |
| 298 | // |
| 299 | // An unknown rule name (a future rule this console has not been taught) is |
| 300 | // rendered as a bare number rather than being given someone else's unit: a |
| 301 | // unitless figure is honest, a wrong unit is not. |
| 302 | func alertValueText(rule string, value float64) string { |
| 303 | switch rule { |
| 304 | case "disk", "mem", "cpu": |
| 305 | return fmt.Sprintf("%.0f%%", value*100) |
| 306 | case "lag": |
| 307 | return fmt.Sprintf("%.0fs", value) |
| 308 | case "geoip": |
| 309 | // A reason code, not a measurement (geoip spec §10.4). |
| 310 | return geoipReasonText(int(value)) |
| 311 | default: |
| 312 | return strconv.FormatFloat(value, 'g', -1, 64) |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | // alertMessage is the operator-facing copy. Plain sentences, and no needle |
| 317 | // beyond the value that triggered the transition — this mail lands in an |
| 318 | // inbox, not on the operator dashboard. |
| 319 | func alertMessage(box, rule string, rs metrics.RuleState, updated time.Time) (subject, body string) { |
| 320 | at := updated |
| 321 | if rs.Firing && !rs.FiredAt.IsZero() { |
| 322 | at = rs.FiredAt |
| 323 | } |
| 324 | stamp := at.UTC().Format(time.RFC3339) |
| 325 | if rs.Firing { |
| 326 | subject = fmt.Sprintf("[carlos] %s: %s alert", box, rule) |
| 327 | body = fmt.Sprintf("The %s alert is firing on %s.\n\nValue at the transition: %s\nTime (UTC): %s\n\n"+ |
| 328 | "You are getting this because you hold the CARLOS fleet-operator role.\n", rule, box, alertValueText(rule, rs.Value), stamp) |
| 329 | return subject, body |
| 330 | } |
| 331 | subject = fmt.Sprintf("[carlos] %s: %s cleared", box, rule) |
| 332 | body = fmt.Sprintf("The %s alert has cleared on %s.\n\nValue at the transition: %s\nTime (UTC): %s\n\n"+ |
| 333 | "You are getting this because you hold the CARLOS fleet-operator role.\n", rule, box, alertValueText(rule, rs.Value), stamp) |
| 334 | return subject, body |
| 335 | } |
| 336 | |
| 337 | // geoipReasonText names a geoip rule reason code (geoip spec §10.4). |
| 338 | func geoipReasonText(code int) string { |
| 339 | switch code { |
| 340 | case 1: |
| 341 | return "database over 45 days old" |
| 342 | case 2: |
| 343 | return "GeoLite2 database expires within 9 days" |
| 344 | case 3: |
| 345 | return "database expired and removed" |
| 346 | case 4: |
| 347 | return "published database not loaded for over an hour" |
| 348 | case 5: |
| 349 | return "feed failing" |
| 350 | } |
| 351 | return "geoip reason " + strconv.Itoa(code) |
| 352 | } |
| 353 | |
amadan