diff --git a/doc/examples/cloudflare-pipelines/.gitignore b/doc/examples/cloudflare-pipelines/.gitignore new file mode 100644 index 0000000000..6e24848eb5 --- /dev/null +++ b/doc/examples/cloudflare-pipelines/.gitignore @@ -0,0 +1,2 @@ +stream-token +stream-url \ No newline at end of file diff --git a/doc/examples/cloudflare-pipelines/README.md b/doc/examples/cloudflare-pipelines/README.md new file mode 100644 index 0000000000..881386b6ca --- /dev/null +++ b/doc/examples/cloudflare-pipelines/README.md @@ -0,0 +1,93 @@ +# Send event recorder events to Cloudflare Pipelines + +The event recorder can send batched JSON events to a Cloudflare Pipelines +stream through its webhook output. + +Create a structured stream matching the event recorder's top-level JSON +envelope: + +```sh +npx wrangler pipelines streams create alertmanager_events \ + --schema-file doc/examples/cloudflare-pipelines/stream-schema.json +``` + +Cloudflare stream schemas and pipeline SQL cannot be modified after creation. +Delete and recreate the stream and pipeline when changing either file. + +Create the R2 bucket and enable R2 Data Catalog on it: + +```sh +npx wrangler r2 bucket create alertmanager-events +npx wrangler r2 bucket catalog enable alertmanager-events +``` + +Create one R2 Data Catalog sink for each destination table. Replace the bucket, +namespace, and catalog token values as needed: + +```sh +for name in lifecycle alerts notifications silences inhibitions; do + npx wrangler pipelines sinks create alertmanager_${name}_sink \ + --type r2-data-catalog \ + --bucket alertmanager-events \ + --namespace alertmanager \ + --table ${name} \ + --catalog-token YOUR_CATALOG_TOKEN \ + --roll-interval 60 +done +``` + +`YOUR_CATALOG_TOKEN` is the value of an R2 API token with **Admin Read & +Write** permission. Create one from **R2 Object Storage** > **Manage API +tokens** > **Create Account API token** in the Cloudflare dashboard. See +[Create an API token](https://developers.cloudflare.com/pipelines/getting-started/#1-create-an-api-token) +in the Cloudflare Pipelines documentation. + +Create a pipeline that fans the stream out to those sinks: + +```sh +npx wrangler pipelines create alertmanager_events_pipeline \ + --sql-file doc/examples/cloudflare-pipelines/pipeline.sql +``` + +The SQL stores process lifecycle, alerts, notifications, silences, and +inhibitions in separate tables. Alert rows promote the `severity`, `service`, +`cluster`, and `team` labels into columns while retaining the complete `labels` +and `annotations` as JSON objects keyed by label name. Each row also includes +commonly queried identifiers and the complete event-specific JSON payload. +Protobuf 64-bit integer fields such as fingerprints, flush IDs, and integration +indexes are strings in protojson and are therefore stored as strings. + +For example, query an arbitrary alert label with R2 SQL using +`json_get_str(labels, 'label_name')`. + +Configure Alertmanager with the stream's HTTP ingestion endpoint: + +```yaml +event_recorder: + webhook_outputs: + - url_file: /etc/alertmanager/cloudflare-stream-url + batch: true + http_config: + authorization: + credentials_file: /etc/alertmanager/cloudflare-stream-token +``` + +The URL file must contain the stream's full ingestion endpoint, such as +`https://.ingest.cloudflare.com`. + +Start Alertmanager with `--enable-feature=event-recorder`. When HTTP ingestion +authentication is enabled, the token must have the `Workers Pipeline Send` +permission. Create one by following +[Create an API token](https://developers.cloudflare.com/fundamentals/api/get-started/create-token/) +in the Cloudflare documentation. This is a different token from the sink catalog +token above: the catalog token lets the sinks write tables to R2, while this one +only lets Alertmanager send events to the stream. + +The event recorder schema encodes labels, annotations, and group labels as JSON +objects keyed by name. The `data` field +remains JSON so the stream accepts every event variant and future additions to +the event recorder schema. The other envelope fields are validated before the +SQL extracts the event-specific protobuf `oneof` from `data`. + +See the Cloudflare documentation for [managing streams](https://developers.cloudflare.com/pipelines/streams/manage-streams/) +and [fan-out pipelines](https://developers.cloudflare.com/pipelines/pipelines/manage-pipelines/#route-one-stream-to-multiple-tables). diff --git a/doc/examples/cloudflare-pipelines/config.yml b/doc/examples/cloudflare-pipelines/config.yml new file mode 100644 index 0000000000..625c9e1084 --- /dev/null +++ b/doc/examples/cloudflare-pipelines/config.yml @@ -0,0 +1,130 @@ +global: + # The smarthost and SMTP sender used for mail notifications. + smtp_smarthost: 'localhost:25' + smtp_from: 'alertmanager@example.org' + smtp_auth_username: 'alertmanager' + smtp_auth_password: 'password' + +# The directory from which notification templates are read. +templates: + - '/etc/alertmanager/template/*.tmpl' + +# The root route on which each incoming alert enters. +route: + # The labels by which incoming alerts are grouped together. For example, + # multiple alerts coming in for cluster=A and alertname=LatencyHigh would + # be batched into a single group. + # + # To aggregate by all possible labels use '...' as the sole label name. + # This effectively disables aggregation entirely, passing through all + # alerts as-is. This is unlikely to be what you want, unless you have + # a very low alert volume or your upstream notification system performs + # its own grouping. Example: group_by: [...] + group_by: ['alertname', 'cluster', 'service'] + + # When a new group of alerts is created by an incoming alert, wait at + # least 'group_wait' to send the initial notification. + # This way ensures that you get multiple alerts for the same group that start + # firing shortly after another are batched together on the first + # notification. + group_wait: 30s + + # When the first notification was sent, wait 'group_interval' to send a batch + # of new alerts that started firing for that group. + group_interval: 5m + + # If an alert has successfully been sent, wait 'repeat_interval' to + # resend them. + repeat_interval: 3h + + # A default receiver + receiver: team-X-mails + + # All the above attributes are inherited by all child routes and can + # overwritten on each. + + # The child route trees. + routes: + # This routes performs a regular expression match on alert labels to + # catch alerts that are related to a list of services. + - matchers: + - service=~"foo1|foo2|baz" + receiver: team-X-mails + # The service has a sub-route for critical alerts, any alerts + # that do not match, i.e. severity != critical, fall-back to the + # parent node and are sent to 'team-X-mails' + routes: + - matchers: + - severity="critical" + receiver: team-X-pager + - matchers: + - service="files" + receiver: team-Y-mails + + routes: + - matchers: + - severity="critical" + receiver: team-Y-pager + + # This route handles all alerts coming from a database service. If there's + # no team to handle it, it defaults to the DB team. + - matchers: + - service="database" + receiver: team-DB-pager + # Also group alerts by affected database. + group_by: [alertname, cluster, database] + routes: + - matchers: + - owner="team-X" + receiver: team-X-pager + continue: true + - matchers: + - owner="team-Y" + receiver: team-Y-pager + + +# Inhibition rules allow to mute a set of alerts given that another alert is +# firing. +# We use this to mute any warning-level notifications if the same alert is +# already critical. +inhibit_rules: + - source_matchers: [severity="critical"] + target_matchers: [severity="warning"] + # Apply inhibition if the alertname is the same. + # CAUTION: + # If all label names listed in `equal` are missing + # from both the source and target alerts, + # the inhibition rule will apply! + equal: [alertname, cluster, service] + + +receivers: + - name: 'team-X-mails' + email_configs: + - to: 'team-X+alerts@example.org' + + - name: 'team-X-pager' + email_configs: + - to: 'team-X+alerts-critical@example.org' + pagerduty_configs: + - service_key: + + - name: 'team-Y-mails' + email_configs: + - to: 'team-Y+alerts@example.org' + + - name: 'team-Y-pager' + pagerduty_configs: + - service_key: + + - name: 'team-DB-pager' + pagerduty_configs: + - service_key: + +event_recorder: + webhook_outputs: + - url_file: ./stream-url + batch: true + http_config: + authorization: + credentials_file: ./stream-token diff --git a/doc/examples/cloudflare-pipelines/pipeline.sql b/doc/examples/cloudflare-pipelines/pipeline.sql new file mode 100644 index 0000000000..14f32fd82f --- /dev/null +++ b/doc/examples/cloudflare-pipelines/pipeline.sql @@ -0,0 +1,170 @@ +INSERT INTO alertmanager_lifecycle_sink +WITH lifecycle_variants AS ( + SELECT + events."@timestamp" AS event_time, + events.instance, + events."clusterPosition" AS cluster_position, + json_get_json(events.data, 'alertmanagerStartupEvent') AS startup, + json_get_json(events.data, 'alertmanagerShutdownEvent') AS shutdown + FROM alertmanager_events AS events +), +lifecycle_events AS ( + SELECT + event_time, + instance, + cluster_position, + CASE WHEN startup IS NOT NULL THEN 'startup' ELSE 'shutdown' END AS event_type, + CASE WHEN startup IS NOT NULL THEN startup ELSE shutdown END AS payload + FROM lifecycle_variants + WHERE startup IS NOT NULL OR shutdown IS NOT NULL +) +SELECT + event_time, + instance, + cluster_position, + event_type, + CASE WHEN event_type = 'startup' THEN json_get_str(payload, 'version') END AS version, + CASE WHEN event_type = 'startup' THEN json_get_str(payload, 'buildContext') END AS build_context, + payload +FROM lifecycle_events; + +INSERT INTO alertmanager_alerts_sink +WITH alert_variants AS ( + SELECT + events."@timestamp" AS event_time, + events.instance, + events."clusterPosition" AS cluster_position, + json_get_json(events.data, 'alertCreated') AS created, + json_get_json(events.data, 'alertResolved') AS resolved, + json_get_json(events.data, 'alertGrouped') AS grouped + FROM alertmanager_events AS events +), +alert_events AS ( + SELECT + event_time, + instance, + cluster_position, + CASE + WHEN created IS NOT NULL THEN 'created' + WHEN resolved IS NOT NULL THEN 'resolved' + ELSE 'grouped' + END AS event_type, + CASE + WHEN created IS NOT NULL THEN json_get_json(created, 'alert') + WHEN resolved IS NOT NULL THEN json_get_json(resolved, 'alert', 'details') + ELSE json_get_json(grouped, 'alert', 'details') + END AS alert, + CASE + WHEN resolved IS NOT NULL THEN json_get_json(resolved, 'groupInfo') + WHEN grouped IS NOT NULL THEN json_get_json(grouped, 'groupInfo') + END AS group_info, + CASE + WHEN created IS NOT NULL THEN created + WHEN resolved IS NOT NULL THEN resolved + ELSE grouped + END AS payload + FROM alert_variants + WHERE created IS NOT NULL OR resolved IS NOT NULL OR grouped IS NOT NULL +) +SELECT + event_time, + instance, + cluster_position, + event_type, + json_get_str(alert, 'name') AS alert_name, + json_get_str(alert, 'fingerprint') AS alert_fingerprint, + json_get_str(group_info, 'groupId') AS group_id, + json_get_str(group_info, 'receiverName') AS receiver_name, + json_get_str(alert, 'labels', 'severity') AS severity, + json_get_str(alert, 'labels', 'service') AS service, + json_get_str(alert, 'labels', 'cluster') AS cluster, + json_get_str(alert, 'labels', 'team') AS team, + json_get_json(alert, 'labels') AS labels, + json_get_json(alert, 'annotations') AS annotations, + payload +FROM alert_events; + +INSERT INTO alertmanager_notifications_sink +WITH notification_events AS ( + SELECT + events."@timestamp" AS event_time, + events.instance, + events."clusterPosition" AS cluster_position, + json_get_json(events.data, 'notification') AS payload + FROM alertmanager_events AS events +) +SELECT + event_time, + instance, + cluster_position, + json_get_str(payload, 'groupInfo', 'groupId') AS group_id, + json_get_str(payload, 'groupInfo', 'receiverName') AS receiver_name, + json_get_str(payload, 'reason') AS reason, + json_get_str(payload, 'integration', 'name') AS integration_name, + json_get_str(payload, 'integration', 'index') AS integration_index, + json_get_str(payload, 'flushId') AS flush_id, + payload +FROM notification_events +WHERE payload IS NOT NULL; + +INSERT INTO alertmanager_silences_sink +WITH silence_variants AS ( + SELECT + events."@timestamp" AS event_time, + events.instance, + events."clusterPosition" AS cluster_position, + json_get_json(events.data, 'silenceCreated') AS created, + json_get_json(events.data, 'silenceUpdated') AS updated, + json_get_json(events.data, 'silenceMutedAlert') AS muted_alert + FROM alertmanager_events AS events +), +silence_events AS ( + SELECT + event_time, + instance, + cluster_position, + CASE + WHEN created IS NOT NULL THEN 'created' + WHEN updated IS NOT NULL THEN 'updated' + ELSE 'muted_alert' + END AS event_type, + CASE + WHEN created IS NOT NULL THEN created + WHEN updated IS NOT NULL THEN updated + ELSE muted_alert + END AS payload + FROM silence_variants + WHERE created IS NOT NULL OR updated IS NOT NULL OR muted_alert IS NOT NULL +) +SELECT + event_time, + instance, + cluster_position, + event_type, + json_get_str(payload, 'silence', 'id') AS silence_id, + json_get_str(payload, 'silence', 'createdBy') AS created_by, + CASE + WHEN event_type = 'muted_alert' THEN json_get_str(payload, 'mutedAlert', 'fingerprint') + END AS muted_alert_fingerprint, + payload +FROM silence_events; + +INSERT INTO alertmanager_inhibitions_sink +WITH inhibition_events AS ( + SELECT + events."@timestamp" AS event_time, + events.instance, + events."clusterPosition" AS cluster_position, + json_get_json(events.data, 'inhibitionMutedAlert') AS payload + FROM alertmanager_events AS events +) +SELECT + event_time, + instance, + cluster_position, + json_get_str(payload, 'mutedAlert', 'fingerprint') AS muted_alert_fingerprint, + json_get_json(payload, 'inhibitRules') AS inhibit_rules, + json_get_json(payload, 'inhibitingFingerprints') AS inhibiting_fingerprints, + payload +FROM inhibition_events +WHERE payload IS NOT NULL; diff --git a/doc/examples/cloudflare-pipelines/stream-schema.json b/doc/examples/cloudflare-pipelines/stream-schema.json new file mode 100644 index 0000000000..3bbbbff10d --- /dev/null +++ b/doc/examples/cloudflare-pipelines/stream-schema.json @@ -0,0 +1,24 @@ +{ + "fields": [ + { + "name": "@timestamp", + "type": "timestamp", + "required": true + }, + { + "name": "instance", + "type": "string", + "required": true + }, + { + "name": "clusterPosition", + "type": "int64", + "required": false + }, + { + "name": "data", + "type": "json", + "required": true + } + ] +} diff --git a/docs/configuration.md b/docs/configuration.md index be9a73f841..0799f0440b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -2188,8 +2188,9 @@ duplicate events after ambiguous failures. With multiple workers, requests may complete out of order; set `workers: 1` when request ordering matters. ```yaml -# URL to POST events to. -url: +# URL to POST events to. Exactly one of url or url_file must be configured. +[ url: ] +[ url_file: ] # HTTP client configuration (TLS, basic auth, OAuth, proxies, ...). [ http_config: ] @@ -2225,19 +2226,8 @@ url: For example, [Cloudflare Pipelines streams](https://developers.cloudflare.com/pipelines/streams/writing-to-streams/) accept JSON arrays through their HTTP ingestion endpoints and can be configured -as a batched webhook output: - -```yaml -event_recorder: - webhook_outputs: - - url: https://.ingest.cloudflare.com - batch: true - http_config: - # The token must have the "Workers Pipeline Send" permission when - # authentication is enabled for the stream. - authorization: - credentials: -``` +as a batched webhook output. See the [Cloudflare Pipelines example](https://github.com/prometheus/alertmanager/tree/main/doc/examples/cloudflare-pipelines) +for a complete setup and configuration. #### `` diff --git a/eventrecorder/webhook.go b/eventrecorder/webhook.go index 1a4972ae1b..366f5ea677 100644 --- a/eventrecorder/webhook.go +++ b/eventrecorder/webhook.go @@ -21,7 +21,9 @@ import ( "log/slog" "net/http" "net/url" + "os" "reflect" + "strings" "sync" "time" @@ -37,7 +39,8 @@ import ( // WebhookOutputConfig configures an HTTP webhook event recorder output. type WebhookOutputConfig struct { // URL is the endpoint to POST each event to. - URL *amcommoncfg.SecretURL `yaml:"url" json:"url"` + URL *amcommoncfg.SecretURL `yaml:"url,omitempty" json:"url,omitempty"` + URLFile string `yaml:"url_file,omitempty" json:"url_file,omitempty"` // HTTPConfig configures the HTTP client used for webhook delivery. HTTPConfig *commoncfg.HTTPClientConfig `yaml:"http_config,omitempty" json:"http_config,omitempty"` // Timeout for webhook HTTP requests (default 10s). @@ -74,10 +77,13 @@ func (c *WebhookOutputConfig) UnmarshalYAML(unmarshal func(any) error) error { if err := unmarshal((*plain)(c)); err != nil { return err } - if c.URL == nil || c.URL.URL == nil { - return errors.New("event_recorder webhook output requires a url") + if c.URL == nil && c.URLFile == "" { + return errors.New("event_recorder webhook output requires one of url or url_file") } - if c.URL.Scheme == "" || c.URL.Host == "" { + if c.URL != nil && c.URLFile != "" { + return errors.New("at most one of url & url_file must be configured") + } + if c.URL != nil && (c.URL.URL == nil || c.URL.Scheme == "" || c.URL.Host == "") { return errors.New("event_recorder webhook output requires an absolute http(s) url") } if c.BatchMaxEvents < 0 || c.BatchMaxBytes < 0 || c.BatchFlushInterval < 0 { @@ -99,7 +105,7 @@ func (c WebhookOutputConfig) equal(o WebhookOutputConfig) bool { if o.URL != nil { bURL = o.URL.String() } - if aURL != bURL { + if aURL != bURL || c.URLFile != o.URLFile { return false } if c.Timeout != o.Timeout { @@ -146,6 +152,7 @@ const ( type WebhookOutput struct { client *http.Client url string + urlFile string name string kind string batch *httpBatchConfig @@ -230,11 +237,16 @@ func newWebhookOutput(cfg WebhookOutputConfig, kind string, batch *httpBatchConf retryBackoff = time.Duration(cfg.RetryBackoff) } - urlStr := cfg.URL.String() - name := fmt.Sprintf("%s:%s", kind, sanitizeURL(urlStr)) + urlStr := "" + name := fmt.Sprintf("%s:url_file:%s", kind, cfg.URLFile) + if cfg.URL != nil { + urlStr = cfg.URL.String() + name = fmt.Sprintf("%s:%s", kind, sanitizeURL(urlStr)) + } wo := &WebhookOutput{ client: client, url: urlStr, + urlFile: cfg.URLFile, name: name, kind: kind, batch: batch, @@ -436,7 +448,19 @@ func (wo *WebhookOutput) postWithRetry(data []byte) { } func (wo *WebhookOutput) post(data []byte) error { - resp, err := wo.client.Post(wo.url, "application/json", bytes.NewReader(data)) + urlStr := wo.url + if wo.urlFile != "" { + content, err := os.ReadFile(wo.urlFile) + if err != nil { + return fmt.Errorf("read event recorder webhook url_file: %w", err) + } + u, err := amcommoncfg.ParseURL(strings.TrimSpace(string(content))) + if err != nil { + return fmt.Errorf("parse event recorder webhook url_file: %w", err) + } + urlStr = u.String() + } + resp, err := wo.client.Post(urlStr, "application/json", bytes.NewReader(data)) if err != nil { return fmt.Errorf("event recorder %s POST failed: %w", wo.kind, err) } diff --git a/eventrecorder/webhook_test.go b/eventrecorder/webhook_test.go index 2cbc79f432..c8a16f083e 100644 --- a/eventrecorder/webhook_test.go +++ b/eventrecorder/webhook_test.go @@ -20,6 +20,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "os" "sync" "sync/atomic" "testing" @@ -95,6 +96,25 @@ func TestWebhookOutput_SendEvent(t *testing.T) { mu.Unlock() } +func TestWebhookOutput_ReadsURLFromFile(t *testing.T) { + var count atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + count.Add(1) + w.WriteHeader(http.StatusNoContent) + })) + defer srv.Close() + + urlFile := t.TempDir() + "/url" + require.NoError(t, os.WriteFile(urlFile, []byte(srv.URL+"\n"), 0o600)) + + wo, err := NewWebhookOutput(WebhookOutputConfig{URLFile: urlFile}, testWebhookDrops(), slog.Default()) + require.NoError(t, err) + _, err = wo.SendEvent(sampleEvent()) + require.NoError(t, err) + require.NoError(t, wo.Close()) + require.Equal(t, int64(1), count.Load()) +} + func TestWebhookOutput_MultipleWorkers(t *testing.T) { var count atomic.Int64 @@ -388,6 +408,19 @@ func TestWebhookOutputConfig_UnmarshalYAML(t *testing.T) { require.Equal(t, "https://example.com/hook", c.URL.String()) }, }, + { + name: "valid url file", + yaml: "url_file: /run/secrets/webhook-url\n", + check: func(t *testing.T, c WebhookOutputConfig) { + require.Nil(t, c.URL) + require.Equal(t, "/run/secrets/webhook-url", c.URLFile) + }, + }, + { + name: "url and url file", + yaml: "url: https://example.com/hook\nurl_file: /run/secrets/webhook-url\n", + wantErr: true, + }, { name: "valid with tunables", yaml: "url: https://example.com/h\ntimeout: 5s\nworkers: 8\nmax_retries: 5\nretry_backoff: 250ms\n",