feat(eventrecorder): add webhook url file + example

doc/examples/cloudflare-pipelines/.gitignore

@@ -0,0 +1,2 @@
+stream-token
+stream-url

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://<stream-id>.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).

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: <team-X-key>
+
+  - name: 'team-Y-mails'
+    email_configs:
+      - to: 'team-Y+alerts@example.org'
+
+  - name: 'team-Y-pager'
+    pagerduty_configs:
+      - service_key: <team-Y-key>
+
+  - name: 'team-DB-pager'
+    pagerduty_configs:
+      - service_key: <team-DB-key>
+
+event_recorder:
+  webhook_outputs:
+  - url_file: ./stream-url
+    batch: true
+    http_config:
+      authorization:
+        credentials_file: ./stream-token

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;