Add native AWS CloudWatch and Azure Monitor alert ingress, plus alert write-back mutations

alert/alert.go

@@ -61,7 +61,7 @@ func (a Alert) Normalize() (*Alert, error) {
 	err := validate.Many(
 		validate.Text("Summary", a.Summary, 1, MaxSummaryLength),
 		validate.Text("Details", a.Details, 0, MaxDetailsLength),
-		validate.OneOf("Source", a.Source, SourceManual, SourceGrafana, SourceSite24x7, SourcePrometheusAlertmanager, SourceEmail, SourceGeneric, SourceUniversal),
+		validate.OneOf("Source", a.Source, SourceManual, SourceGrafana, SourceSite24x7, SourcePrometheusAlertmanager, SourceCloudwatch, SourceAzureMonitor, SourceEmail, SourceGeneric, SourceUniversal),
 		validate.OneOf("Status", a.Status, StatusTriggered, StatusActive, StatusClosed),
 		validate.UUID("ServiceID", a.ServiceID),
 	)

alert/alertlog/legacylogs.go

@@ -103,6 +103,10 @@ func createdSubject(msg string) *Subject {
 		return &Subject{Type: SubjectTypeIntegrationKey, Classifier: "Site24x7"}
 	case "Created via: prometheusAlertmanager":
 		return &Subject{Type: SubjectTypeIntegrationKey, Classifier: "PrometheusAlertmanager"}
+	case "Created via: cloudwatch":
+		return &Subject{Type: SubjectTypeIntegrationKey, Classifier: "CloudWatch"}
+	case "Created via: azureMonitor":
+		return &Subject{Type: SubjectTypeIntegrationKey, Classifier: "Azure Monitor"}
 	case "Created via: manual":
 		return &Subject{Type: SubjectTypeUser, Classifier: "Web"}
 	case "Created via: generic":

alert/alertlog/store.go

@@ -384,6 +384,10 @@ func (s *Store) logEntry(ctx context.Context, tx *sql.Tx, _type Type, meta inter
 				r.subject.classifier = "Grafana"
 			case integrationkey.TypeSite24x7:
 				r.subject.classifier = "Site24x7"
+			case integrationkey.TypeCloudwatch:
+				r.subject.classifier = "CloudWatch"
+			case integrationkey.TypeAzureMonitor:
+				r.subject.classifier = "Azure Monitor"
 			case integrationkey.TypeEmail:
 				r.subject.classifier = "Email"
 			}

alert/details.go

@@ -0,0 +1,61 @@
+package alert
+
+import (
+	"context"
+
+	"github.com/target/goalert/gadb"
+	"github.com/target/goalert/permission"
+	"github.com/target/goalert/validation/validate"
+)
+
+// LockDetailsTx returns an alert's current Details, locking its row for the rest
+// of the transaction.
+//
+// This exists so an append can be done as a read-modify-write without losing a
+// concurrent one: the lock makes a second appender wait for the first to commit
+// and then read the value it wrote, instead of both reading the same original and
+// one overwriting the other.
+func (s Store) LockDetailsTx(ctx context.Context, db gadb.DBTX, alertID int) (string, error) {
+	err := permission.LimitCheckAny(ctx, permission.User, permission.Service)
+	if err != nil {
+		return "", err
+	}
+
+	return gadb.New(db).Alert_LockOneAlertDetails(ctx, gadb.Alert_LockOneAlertDetailsParams{
+		ID:        int64(alertID),
+		ServiceID: permission.ServiceNullUUID(ctx), // only provide service_id restriction if request is from a service
+	})
+}
+
+// SetDetailsTx sets the Details for an existing alert, replacing it wholesale.
+//
+// Unlike SetMetadataTx this does not merge: Details is free text with no
+// key/value structure to merge on, so a caller wanting to preserve existing
+// content must read it first and pass the full text it wants stored.
+func (s Store) SetDetailsTx(ctx context.Context, db gadb.DBTX, alertID int, details string) error {
+	err := permission.LimitCheckAny(ctx, permission.User, permission.Service)
+	if err != nil {
+		return err
+	}
+
+	err = validate.Text("Details", details, 0, MaxDetailsLength)
+	if err != nil {
+		return err
+	}
+
+	rowCount, err := gadb.New(db).Alert_SetDetails(ctx, gadb.Alert_SetDetailsParams{
+		ID:        int64(alertID),
+		Details:   details,
+		ServiceID: permission.ServiceNullUUID(ctx), // only provide service_id restriction if request is from a service
+	})
+	if err != nil {
+		return err
+	}
+
+	if rowCount == 0 {
+		// shouldn't happen, but just in case
+		return permission.NewAccessDenied("alert closed, invalid, or wrong service")
+	}
+
+	return nil
+}

alert/metadata.go

@@ -20,6 +20,22 @@ type metadataDBFormat struct {
 	AlertMetaV1 map[string]string
 }
 
+// LockMetadataTx locks the alert's row for the remainder of the transaction, so
+// a metadata read-modify-write cannot interleave with a concurrent one. Returns
+// sql.ErrNoRows if the alert does not exist.
+//
+// Callers must take this lock BEFORE calling Metadata, or the lock does nothing:
+// locking after the read has already raced.
+func (s Store) LockMetadataTx(ctx context.Context, db gadb.DBTX, alertID int) error {
+	err := permission.LimitCheckAny(ctx, permission.System, permission.User, permission.Service)
+	if err != nil {
+		return err
+	}
+
+	_, err = gadb.New(db).Alert_LockOneAlertMetadata(ctx, int64(alertID))
+	return err
+}
+
 // Metadata returns the metadata for a single alert. If err == nil, meta is guaranteed to be non-nil. If the alert has no metadata, an empty map is returned.
 func (s *Store) Metadata(ctx context.Context, db gadb.DBTX, alertID int) (meta map[string]string, err error) {
 	err = permission.LimitCheckAny(ctx, permission.System, permission.User)

alert/queries.sql

@@ -104,6 +104,48 @@ ON CONFLICT (alert_id)
     WHERE
         alert_data.alert_id = $1;
 
+-- name: Alert_LockOneAlertMetadata :one
+-- Locks the alert's row for a metadata read-modify-write, so two concurrent
+-- writers cannot both read the same starting document and one silently
+-- overwrite the other. Locks alerts, not alert_data: alert_data has no row
+-- before an alert's first metadata write, and FOR UPDATE against a table with
+-- no matching row locks nothing, so it could not serialize the first-writer
+-- case, which is the common one for a brand new alert.
+SELECT
+    id
+FROM
+    alerts
+WHERE
+    id = $1
+FOR UPDATE;
+
+-- name: Alert_LockOneAlertDetails :one
+-- Returns the details for the alert and locks its row, so that a read-modify-write
+-- of details cannot interleave with a concurrent one.
+SELECT
+    details
+FROM
+    alerts
+WHERE
+    id = $1
+    -- ensure the alert is associated with the service, if coming from an integration
+    AND (service_id = $2
+        OR $2 IS NULL)
+FOR UPDATE;
+
+-- name: Alert_SetDetails :execrows
+-- Sets the details for the alert.
+UPDATE
+    alerts
+SET
+    details = $2
+WHERE
+    id = $1
+    AND status != 'closed'
+    -- ensure the alert is associated with the service, if coming from an integration
+    AND (service_id = $3
+        OR $3 IS NULL);
+
 -- name: Alert_ServiceEPHasSteps :one
 -- Returns true if the Escalation Policy for the provided service has at least one step.
 SELECT

alert/source.go

@@ -14,6 +14,8 @@ const (
 	SourceGrafana                Source = "grafana"                // grafana alert
 	SourceSite24x7               Source = "site24x7"               // site24x7 alert
 	SourcePrometheusAlertmanager Source = "prometheusAlertmanager" // prometheus alertmanager alert
+	SourceCloudwatch             Source = "cloudwatch"             // AWS CloudWatch alarm (via SNS)
+	SourceAzureMonitor           Source = "azureMonitor"           // Azure Monitor alert (action group webhook)
 	SourceManual                 Source = "manual"                 // manually triggered
 	SourceGeneric                Source = "generic"                // generic API
 	SourceUniversal              Source = "universal"              // universal integration

app/config.go

@@ -61,6 +61,11 @@ type Config struct {
 	TwilioBaseURL string
 	SlackBaseURL  string
 
+	// CloudwatchBaseURL overrides the origin used for outbound SNS certificate
+	// and subscription-confirmation requests. Testing only; there is no flag for
+	// it. The host allowlist still runs against the message-supplied URL.
+	CloudwatchBaseURL string
+
 	DBURL     string
 	DBURLNext string
 

app/inithttp.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/target/goalert/app/csp"
+	"github.com/target/goalert/azuremonitor"
+	"github.com/target/goalert/cloudwatch"
 	"github.com/target/goalert/config"
 	"github.com/target/goalert/expflag"
 	"github.com/target/goalert/genericapi"
@@ -122,6 +124,15 @@ func (app *App) initHTTP(ctx context.Context) error {
 		UserStore:           app.UserStore,
 	})
 
+	cw, err := cloudwatch.NewHandler(cloudwatch.Config{
+		AlertStore:          app.AlertStore,
+		IntegrationKeyStore: app.IntegrationKeyStore,
+		BaseURL:             app.cfg.CloudwatchBaseURL,
+	})
+	if err != nil {
+		return err
+	}
+
 	mux.Handle("POST /api/graphql", app.graphql2.Handler())
 
 	mux.HandleFunc("GET /api/v2/config", app.ConfigStore.ServeConfig)
@@ -148,6 +159,11 @@ func (app *App) initHTTP(ctx context.Context) error {
 	mux.HandleFunc("POST /api/v2/grafana/incoming", grafana.GrafanaToEventsAPI(app.AlertStore, app.IntegrationKeyStore))
 	mux.HandleFunc("POST /api/v2/site24x7/incoming", site24x7.Site24x7ToEventsAPI(app.AlertStore, app.IntegrationKeyStore))
 	mux.HandleFunc("POST /api/v2/prometheusalertmanager/incoming", prometheus.PrometheusAlertmanagerEventsAPI(app.AlertStore, app.IntegrationKeyStore))
+	mux.HandleFunc("POST /api/v2/cloudwatch/incoming", cw.ServeIncoming)
+	mux.HandleFunc("POST /api/v2/azuremonitor/incoming", azuremonitor.NewHandler(azuremonitor.Config{
+		AlertStore:          app.AlertStore,
+		IntegrationKeyStore: app.IntegrationKeyStore,
+	}).ServeIncoming)
 
 	mux.HandleFunc("POST /api/v2/generic/incoming", generic.ServeCreateAlert)
 	mux.HandleFunc("POST /api/v2/heartbeat/{heartbeatID}", generic.ServeHeartbeatCheck)

auth/handler.go

@@ -594,6 +594,10 @@ func (h *Handler) authWithToken(w http.ResponseWriter, req *http.Request, next h
 		ctx, err = h.cfg.IntKeyStore.Authorize(ctx, *tok, integrationkey.TypeSite24x7)
 	case "/api/v2/prometheusalertmanager/incoming":
 		ctx, err = h.cfg.IntKeyStore.Authorize(ctx, *tok, integrationkey.TypePrometheusAlertmanager)
+	case "/api/v2/cloudwatch/incoming":
+		ctx, err = h.cfg.IntKeyStore.Authorize(ctx, *tok, integrationkey.TypeCloudwatch)
+	case "/api/v2/azuremonitor/incoming":
+		ctx, err = h.cfg.IntKeyStore.Authorize(ctx, *tok, integrationkey.TypeAzureMonitor)
 	case "/api/v2/calendar":
 		ctx, err = h.cfg.CalSubStore.Authorize(ctx, *tok)
 	default:

azuremonitor/azuremonitor.go

@@ -0,0 +1,140 @@
+// Package azuremonitor implements ingress for Azure Monitor alerts delivered by
+// an action-group webhook receiver.
+//
+// # Trust model
+//
+// Azure signs nothing. Unlike the sibling cloudwatch integration -- where SNS
+// signs every message with a verifiable certificate chain -- the integration key
+// in the query string is the only credential, so the webhook URL is
+// credential-grade: anyone holding it can create arbitrary alerts on the service
+// the key identifies. There is no public key to verify against, so asymmetric
+// verification is not possible; anything built would be a shared secret with
+// extra steps.
+//
+// The handler makes no outbound requests at all, so there is no SSRF surface and
+// nothing analogous to cloudwatch's host allowlist or certificate cache.
+//
+// # Service resolution
+//
+// The integration key selects the GoAlert service. Nothing here knows or can know
+// which action group, subscription, or team it is serving -- the only inputs are
+// the request body and the key. That is what lets one deployed handler serve any
+// number of action groups with no routing table and no per-group code.
+package azuremonitor
+
+import (
+	"errors"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/target/goalert/alert"
+	"github.com/target/goalert/integrationkey"
+	"github.com/target/goalert/permission"
+	"github.com/target/goalert/retry"
+	"github.com/target/goalert/util/errutil"
+	"github.com/target/goalert/util/log"
+)
+
+// maxBodyBytes bounds the request body independently of the global
+// maxBodySizeMiddleware, which is disabled when MaxReqBodyBytes is 0.
+const maxBodyBytes = 256 * 1024
+
+// Config configures a Handler.
+type Config struct {
+	AlertStore          *alert.Store
+	IntegrationKeyStore *integrationkey.Store
+}
+
+// Handler serves the Azure Monitor ingress endpoint.
+type Handler struct {
+	cfg Config
+}
+
+// NewHandler returns a Handler for the given config.
+func NewHandler(cfg Config) *Handler { return &Handler{cfg: cfg} }
+
+// ServeIncoming handles an Azure Monitor action-group webhook delivery.
+func (h *Handler) ServeIncoming(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	err := permission.LimitCheckAny(ctx, permission.Service)
+	if errutil.HTTPError(ctx, w, err) {
+		return
+	}
+
+	// MaxBytesReader rather than io.LimitReader: errutil.HTTPError maps
+	// *http.MaxBytesError to a clean 413, whereas silent truncation would surface
+	// as a confusing 400.
+	r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes)
+	data, err := io.ReadAll(r.Body)
+	if errutil.HTTPErrorRetry(ctx, w, err) {
+		return
+	}
+
+	a, meta, info, err := buildAlert(data)
+	switch {
+	case errors.Is(err, errLegacySchema):
+		// Deliberately not degraded to the best-effort path: a legacy-schema
+		// payload would yield alerts with no useful content and no indication
+		// why. 400 rather than 5xx so Azure does not retry a receiver that is
+		// permanently misconfigured.
+		log.Logf(ctx, "azuremonitor: %v", err)
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	case err != nil:
+		log.Debugf(ctx, "azuremonitor: bad request body: %v", err)
+		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+		return
+	}
+
+	// signalType alone does not identify the payload shape -- Platform and
+	// Prometheus metric alerts share it, as do Log Alerts V2 and Azure Backup --
+	// so log the whole triple that actually determines parsing.
+	ctx = log.WithFields(ctx, log.Fields{
+		"AzureSignalType":       info.SignalType,
+		"AzureMonitorService":   info.MonitorService,
+		"AzureConditionType":    info.ConditionType,
+		"AzureMonitorCondition": meta["monitor_condition"],
+	})
+
+	// Nothing recognised alertContext, so this alert carries essentials only. That
+	// is a deliberate graceful degradation rather than a failure, but it is also
+	// how a newly-routed Azure alert type announces itself -- otherwise the only
+	// symptom is thin alerts that nobody notices.
+	if !info.ContextRendered {
+		log.Logf(ctx, "azuremonitor: unrecognised alertContext, alert built from essentials only")
+	}
+
+	a.ServiceID = permission.ServiceID(ctx)
+	if len(meta) == 0 {
+		meta = nil
+	}
+
+	var created *alert.Alert
+	err = retry.DoTemporaryError(func(int) error {
+		var err error
+		created, _, err = h.cfg.AlertStore.CreateOrUpdateWithMeta(ctx, &a, meta)
+		return err
+	},
+		retry.Log(ctx),
+		retry.Limit(5),
+		retry.FibBackoff(250*time.Millisecond),
+	)
+	// HTTPErrorRetry, not HTTPError: Azure's retryable set is 408, 429, 503 and 504
+	// -- 500 is absent, so mapping an exhausted database failure to 500 would drop
+	// the alert with no retries at all, which is exactly the case most worth
+	// retrying (an Aurora failover). 503 is retried by Azure and by SNS.
+	if errutil.HTTPErrorRetry(ctx, w, err) {
+		return
+	}
+
+	// created is nil with a nil error when the status was closed and no open alert
+	// held this dedup key -- a Resolved delivery for an alert we never opened, or
+	// already closed. Normal, so info level and a 2xx.
+	if created == nil {
+		log.Logf(ctx, "azuremonitor: no open alert to close")
+	}
+
+	w.WriteHeader(http.StatusNoContent)
+}