| 1 | // Package alerting is the operator alert delivery the agent and the console |
| 2 | // can share: a webhook, an SNS topic, either, both or neither (Sinks), and |
| 3 | // Page, the paging primitive the moves code calls (moves spec §13). |
| 4 | // |
| 5 | // Delivery is best effort. A sink that fails is counted through its failure |
| 6 | // hook and its error returned; the other sink still runs. Nothing here |
| 7 | // retries beyond what the AWS SDK does inside one publish. |
| 8 | package alerting |
| 9 | |
| 10 | import ( |
| 11 | "bytes" |
| 12 | "context" |
| 13 | "encoding/json" |
| 14 | "errors" |
| 15 | "fmt" |
| 16 | "log" |
| 17 | "net/http" |
| 18 | "regexp" |
| 19 | "strings" |
| 20 | "sync" |
| 21 | "sync/atomic" |
| 22 | "time" |
| 23 | |
| 24 | "github.com/aws/aws-sdk-go-v2/aws" |
| 25 | "github.com/aws/aws-sdk-go-v2/aws/arn" |
| 26 | awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| 27 | "github.com/aws/aws-sdk-go-v2/service/sns" |
| 28 | snstypes "github.com/aws/aws-sdk-go-v2/service/sns/types" |
| 29 | "github.com/carlosframework/platform/internal/rootaws" |
| 30 | ) |
| 31 | |
| 32 | // Page levels (moves spec §13): an incident wants someone now, a warning |
| 33 | // wants someone today. |
| 34 | const ( |
| 35 | LevelIncident = "incident" |
| 36 | LevelWarning = "warning" |
| 37 | ) |
| 38 | |
| 39 | // Publisher is the one SNS call this package makes. |
| 40 | type Publisher interface { |
| 41 | Publish(context.Context, *sns.PublishInput, ...func(*sns.Options)) (*sns.PublishOutput, error) |
| 42 | } |
| 43 | |
| 44 | // PublishTimeout bounds one SNS publish, so a black-holed endpoint cannot |
| 45 | // hold the caller for the SDK's own default retries. |
| 46 | const PublishTimeout = 10 * time.Second |
| 47 | |
| 48 | // subjectLimit is SNS's: a subject is under 100 characters. |
| 49 | const subjectLimit = 100 |
| 50 | |
| 51 | // Sinks is one process's delivery. An empty WebhookURL or a nil Post |
| 52 | // switches the webhook off; an empty TopicARN or a nil Pub switches SNS off. |
| 53 | // The failure hooks count a failed send; nil counts nothing. |
| 54 | type Sinks struct { |
| 55 | WebhookURL string |
| 56 | Post func(url string, body []byte) error |
| 57 | TopicARN string |
| 58 | Pub Publisher |
| 59 | |
| 60 | WebhookFailed func() |
| 61 | SNSFailed func() |
| 62 | } |
| 63 | |
| 64 | // HasWebhook and HasSNS say whether that sink is on. Safe on nil. |
| 65 | func (s *Sinks) HasWebhook() bool { return s != nil && s.WebhookURL != "" && s.Post != nil } |
| 66 | func (s *Sinks) HasSNS() bool { return s != nil && s.TopicARN != "" && s.Pub != nil } |
| 67 | |
| 68 | // PostWebhook posts body to the webhook. A failure is counted and returned. |
| 69 | // The caller checks HasWebhook first. |
| 70 | func (s *Sinks) PostWebhook(body []byte) error { |
| 71 | if err := s.Post(s.WebhookURL, body); err != nil { |
| 72 | if s.WebhookFailed != nil { |
| 73 | s.WebhookFailed() |
| 74 | } |
| 75 | return err |
| 76 | } |
| 77 | return nil |
| 78 | } |
| 79 | |
| 80 | // PublishSNS publishes a subject and message to the topic, bounded by |
| 81 | // PublishTimeout. A failure is counted and returned. The caller checks |
| 82 | // HasSNS first. |
| 83 | func (s *Sinks) PublishSNS(ctx context.Context, subject, message string) error { |
| 84 | return s.publish(ctx, &sns.PublishInput{TopicArn: &s.TopicARN, Subject: &subject, Message: &message}) |
| 85 | } |
| 86 | |
| 87 | func (s *Sinks) publish(ctx context.Context, in *sns.PublishInput) error { |
| 88 | pctx, cancel := context.WithTimeout(ctx, PublishTimeout) |
| 89 | defer cancel() |
| 90 | if _, err := s.Pub.Publish(pctx, in); err != nil { |
| 91 | if s.SNSFailed != nil { |
| 92 | s.SNSFailed() |
| 93 | } |
| 94 | return err |
| 95 | } |
| 96 | return nil |
| 97 | } |
| 98 | |
| 99 | // pagePayload is a page on the webhook. |
| 100 | type pagePayload struct { |
| 101 | Kind string `json:"kind"` |
| 102 | Level string `json:"level"` |
| 103 | Subject string `json:"subject"` |
| 104 | Body string `json:"body"` |
| 105 | } |
| 106 | |
| 107 | // Page sends one page to every configured sink: the webhook gets |
| 108 | // {"kind":"page","level","subject","body"}, SNS the subject (made to fit |
| 109 | // SNS's rules) and the body, with the level as a message attribute so a |
| 110 | // subscription can filter on it. The caller's subject and body are the |
| 111 | // whole message: no URL, ARN or credential is added here, and the caller |
| 112 | // must not put a secret or a tenant path in either. With no sink it logs |
| 113 | // once (until the next Arm) and returns nil. A failed sink is counted, and |
| 114 | // every failure is returned joined. |
| 115 | func (s *Sinks) Page(ctx context.Context, level, subject, body string) error { |
| 116 | if !s.HasWebhook() && !s.HasSNS() { |
| 117 | if !noSinkLogged.Swap(true) { |
| 118 | log.Printf("alerting: page not delivered, no sink configured (set CARLOS_ALERT_SNS_TOPIC_ARN or CARLOS_ALERT_WEBHOOK): %s", fitSubject(subject)) |
| 119 | } |
| 120 | return nil |
| 121 | } |
| 122 | var errs []error |
| 123 | if s.HasWebhook() { |
| 124 | b, err := json.Marshal(pagePayload{Kind: "page", Level: level, Subject: subject, Body: body}) |
| 125 | if err == nil { |
| 126 | err = s.PostWebhook(b) |
| 127 | } |
| 128 | if err != nil { |
| 129 | errs = append(errs, fmt.Errorf("webhook: %w", err)) |
| 130 | } |
| 131 | } |
| 132 | if s.HasSNS() { |
| 133 | subj := fitSubject(subject) |
| 134 | in := &sns.PublishInput{TopicArn: &s.TopicARN, Subject: &subj, Message: &body, |
| 135 | MessageAttributes: map[string]snstypes.MessageAttributeValue{"level": {DataType: aws.String("String"), StringValue: aws.String(level)}}} |
| 136 | if err := s.publish(ctx, in); err != nil { |
| 137 | errs = append(errs, fmt.Errorf("sns: %w", err)) |
| 138 | } |
| 139 | } |
| 140 | return errors.Join(errs...) |
| 141 | } |
| 142 | |
| 143 | // fitSubject makes a subject SNS accepts: printable ASCII, no line break, |
| 144 | // under 100 characters. A line break becomes a space and anything else |
| 145 | // outside printable ASCII a question mark. |
| 146 | func fitSubject(s string) string { |
| 147 | out := make([]byte, 0, len(s)) |
| 148 | for _, r := range s { |
| 149 | switch { |
| 150 | case r == '\n' || r == '\r' || r == '\t': |
| 151 | if n := len(out); n == 0 || out[n-1] != ' ' { |
| 152 | out = append(out, ' ') |
| 153 | } |
| 154 | case r < 0x20 || r > 0x7e: |
| 155 | out = append(out, '?') |
| 156 | default: |
| 157 | out = append(out, byte(r)) |
| 158 | } |
| 159 | if len(out) >= subjectLimit-1 { |
| 160 | break |
| 161 | } |
| 162 | } |
| 163 | return string(out) |
| 164 | } |
| 165 | |
| 166 | // The armed sinks: the process's own, set once its sinks are built (Arm). |
| 167 | var ( |
| 168 | armedMu sync.RWMutex |
| 169 | armed *Sinks |
| 170 | noSinkLogged atomic.Bool |
| 171 | ) |
| 172 | |
| 173 | // Arm makes s the sinks Page delivers on. nil means none. Arming starts a |
| 174 | // fresh "no sink" notice. |
| 175 | func Arm(s *Sinks) { |
| 176 | armedMu.Lock() |
| 177 | armed = s |
| 178 | armedMu.Unlock() |
| 179 | noSinkLogged.Store(false) |
| 180 | } |
| 181 | |
| 182 | // Page pages an operator on the armed sinks (Sinks.Page). |
| 183 | func Page(ctx context.Context, level, subject, body string) error { |
| 184 | armedMu.RLock() |
| 185 | s := armed |
| 186 | armedMu.RUnlock() |
| 187 | return s.Page(ctx, level, subject, body) |
| 188 | } |
| 189 | |
| 190 | // FromEnv builds a process's sinks from the agent's two variables, |
| 191 | // CARLOS_ALERT_WEBHOOK (posted through PostJSON) and |
| 192 | // CARLOS_ALERT_SNS_TOPIC_ARN (published from the topic's own region through |
| 193 | // newClient, NewSNSClient when nil). The console arms these for its move |
| 194 | // pass (moves Plan 5 Task 7). A topic that does not parse, or whose client |
| 195 | // cannot be built, leaves SNS off and says so; the webhook is unaffected. |
| 196 | // It never returns nil: with neither set, Page logs that no sink is |
| 197 | // configured. |
| 198 | func FromEnv(ctx context.Context, getenv func(string) string, newClient func(context.Context, string) (Publisher, error)) *Sinks { |
| 199 | if newClient == nil { |
| 200 | newClient = NewSNSClient |
| 201 | } |
| 202 | s := &Sinks{} |
| 203 | if url := strings.TrimSpace(getenv("CARLOS_ALERT_WEBHOOK")); url != "" { |
| 204 | s.WebhookURL, s.Post = url, PostJSON |
| 205 | } |
| 206 | if topic := strings.TrimSpace(getenv("CARLOS_ALERT_SNS_TOPIC_ARN")); topic != "" { |
| 207 | region, err := ParseTopicARN(topic) |
| 208 | if err != nil { |
| 209 | log.Printf("alerting: CARLOS_ALERT_SNS_TOPIC_ARN is not usable, SNS sink off: %v", err) |
| 210 | return s |
| 211 | } |
| 212 | pub, err := newClient(ctx, region) |
| 213 | if err != nil { |
| 214 | log.Printf("alerting: build the SNS client, SNS sink off: %v", err) |
| 215 | return s |
| 216 | } |
| 217 | s.TopicARN, s.Pub = topic, pub |
| 218 | } |
| 219 | return s |
| 220 | } |
| 221 | |
| 222 | // PostJSON is the production webhook post: a 5 second POST with a JSON |
| 223 | // content type. Any non-2xx status is an error. |
| 224 | func PostJSON(url string, body []byte) error { |
| 225 | client := &http.Client{Timeout: 5 * time.Second} |
| 226 | resp, err := client.Post(url, "application/json", bytes.NewReader(body)) |
| 227 | if err != nil { |
| 228 | return err |
| 229 | } |
| 230 | defer resp.Body.Close() |
| 231 | if resp.StatusCode < 200 || resp.StatusCode >= 300 { |
| 232 | return fmt.Errorf("host: webhook: non-2xx status %s", resp.Status) |
| 233 | } |
| 234 | return nil |
| 235 | } |
| 236 | |
| 237 | // topicARNPattern is the whole shape of a standard SNS topic ARN: partition, |
| 238 | // region, a 12 digit account, and a topic name that is only letters, digits, |
| 239 | // hyphen and underscore. The name has no colon, so a subscription ARN |
| 240 | // (topic ARN plus ":<subscription-id>") does not match, and no dot, so FIFO |
| 241 | // topics (which need a message group id this sink does not send) do not |
| 242 | // either. |
| 243 | var topicARNPattern = regexp.MustCompile(`^arn:aws(-cn|-us-gov)?:sns:([a-z0-9-]+):[0-9]{12}:[A-Za-z0-9_-]{1,256}$`) |
| 244 | |
| 245 | // ParseTopicARN checks a CARLOS_ALERT_SNS_TOPIC_ARN value and returns the |
| 246 | // region the topic lives in. SNS is regional and a topic is reached only |
| 247 | // through its own region's endpoint, so the region is read from the ARN and |
| 248 | // not from AWS_REGION (host.env pins that to the deployment bucket's region |
| 249 | // on every flagship box, which would aim three of the four at the wrong |
| 250 | // place). |
| 251 | func ParseTopicARN(s string) (region string, err error) { |
| 252 | s = strings.TrimSpace(s) |
| 253 | if s == "" { |
| 254 | return "", errors.New("empty") |
| 255 | } |
| 256 | a, err := arn.Parse(s) |
| 257 | if err != nil { |
| 258 | return "", err |
| 259 | } |
| 260 | if a.Service != "sns" { |
| 261 | return "", fmt.Errorf("service %q, want sns", a.Service) |
| 262 | } |
| 263 | if !topicARNPattern.MatchString(s) { |
| 264 | return "", errors.New("not a topic arn (want arn:aws:sns:<region>:<12 digit account>:<topic name>)") |
| 265 | } |
| 266 | return a.Region, nil |
| 267 | } |
| 268 | |
| 269 | // NewSNSClient is the production publisher: the default credential chain |
| 270 | // (a box's instance role, the console's task role), aimed at the topic's |
| 271 | // own region. |
| 272 | func NewSNSClient(ctx context.Context, region string) (Publisher, error) { |
| 273 | cfg, err := rootaws.LoadDefaultConfig(ctx, awsconfig.WithRegion(region)) |
| 274 | if err != nil { |
| 275 | return nil, err |
| 276 | } |
| 277 | return publishFunc(sns.NewFromConfig(cfg).Publish), nil |
| 278 | } |
| 279 | |
| 280 | // publishFunc is the Publisher over one bound method value. It is not |
| 281 | // `return sns.NewFromConfig(cfg), nil` on purpose: an *sns.Client stored in |
| 282 | // an interface makes the linker keep every exported SNS method once |
| 283 | // html/template is linked in (discussion #275, ~0.75 MB here and ~15 MB |
| 284 | // across the SDK clients). A method value keeps Publish alone. |
| 285 | type publishFunc func(context.Context, *sns.PublishInput, ...func(*sns.Options)) (*sns.PublishOutput, error) |
| 286 | |
| 287 | func (f publishFunc) Publish(ctx context.Context, in *sns.PublishInput, opts ...func(*sns.Options)) (*sns.PublishOutput, error) { |
| 288 | return f(ctx, in, opts...) |
| 289 | } |
| 290 | |
amadan