Postmortem: A Misconfigured Sentry 8.0 Alert Caused Missed Production Error in Our Payment App

import os
import sys
import json
import logging
from typing import Dict, Any, Optional
from sentry_sdk import client as sentry_client
from sentry_api import SentryApiClient  # Internal wrapper for Sentry 8.0 REST API

# Configure logging for audit trail
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class SentryAlertConfigurator:
    """Handles creation and validation of Sentry 8.0 alert rules for payment services."""

    def __init__(self, org_slug: str, project_slug: str):
        self.api_token = os.getenv("SENTRY_AUTH_TOKEN")
        if not self.api_token:
            logger.error("Missing SENTRY_AUTH_TOKEN environment variable")
            sys.exit(1)
        self.client = SentryApiClient(
            base_url="https://sentry.io/api/0",
            auth_token=self.api_token
        )
        self.org_slug = org_slug
        self.project_slug = project_slug
        self.project_id = self._fetch_project_id()

    def _fetch_project_id(self) -> str:
        """Retrieve Sentry project ID for the given org/project slugs."""
        try:
            resp = self.client.get(f"/organizations/{self.org_slug}/projects/")
            resp.raise_for_status()
            projects = resp.json()
            for proj in projects:
                if proj["slug"] == self.project_slug:
                    logger.info(f"Found project ID: {proj['id']} for {self.project_slug}")
                    return proj["id"]
            logger.error(f"Project {self.project_slug} not found in org {self.org_slug}")
            sys.exit(1)
        except Exception as e:
            logger.error(f"Failed to fetch project ID: {str(e)}")
            sys.exit(1)

    def create_payment_error_alert(self) -> Optional[Dict[str, Any]]:
        """
        Create Sentry 8.0 alert rule for payment processing errors.
        NOTE: This is the misconfigured version that caused the incident.
        """
        alert_payload = {
            "name": "Payment Service Error Alert",
            "conditions": [
                {
                    "id": "sentry.rules.conditions.first_seen_event.FirstSeenEventCondition",
                    "firstSeenWindow": 300  # 5 minute aggregation window (default, too short)
                },
                {
                    "id": "sentry.rules.conditions.event_frequency.EventFrequencyCondition",
                    "frequencyWindow": 300,
                    "value": 10,  # Only alert if 10+ events in 5 minutes
                    "comparisonType": "count"
                }
            ],
            "filters": [
                {
                    "id": "sentry.rules.filters.tagged_event.TaggedEventFilter",
                    "key": "service",
                    "value": "payment-processor"
                },
                {
                    "id": "sentry.rules.filters.issue_category.IssueCategoryFilter",
                    "value": ["error"]
                }
            ],
            "actions": [
                {
                    "id": "sentry.rules.actions.notify_event.NotifyEventAction"
                }
            ],
            "actionMatch": "all",
            "frequency": 30  # Alert at most every 30 minutes
        }

        try:
            resp = self.client.post(
                f"/projects/{self.org_slug}/{self.project_slug}/rules/",
                json=alert_payload
            )
            resp.raise_for_status()
            rule = resp.json()
            logger.info(f"Created alert rule: {rule['id']} - {rule['name']}")
            return rule
        except Exception as e:
            logger.error(f"Failed to create alert rule: {str(e)}")
            logger.error(f"Response body: {getattr(e, 'response', {}).get('text', 'N/A')}")
            return None

if __name__ == "__main__":
    # Production config values used during the incident
    configurator = SentryAlertConfigurator(
        org_slug="fintech-prod-org",
        project_slug="payment-processor-v2"
    )
    rule = configurator.create_payment_error_alert()
    if not rule:
        sys.exit(1)
    print(json.dumps(rule, indent=2))

import os
import sys
import json
import logging
from typing import Dict, Any, Optional, List
from sentry_api import SentryApiClient

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class FixedSentryAlertConfigurator:
    """Deploys corrected Sentry 8.0 alert rules for low-frequency critical services."""

    def __init__(self, org_slug: str, project_slug: str):
        self.api_token = os.getenv("SENTRY_AUTH_TOKEN")
        if not self.api_token:
            logger.error("Missing SENTRY_AUTH_TOKEN")
            sys.exit(1)
        self.client = SentryApiClient(
            base_url="https://sentry.io/api/0",
            auth_token=self.api_token
        )
        self.org_slug = org_slug
        self.project_slug = project_slug

    def delete_legacy_rules(self, rule_prefix: str) -> List[str]:
        """Delete all legacy alert rules with the given prefix to avoid conflicts."""
        deleted_ids = []
        try:
            resp = self.client.get(f"/projects/{self.org_slug}/{self.project_slug}/rules/")
            resp.raise_for_status()
            rules = resp.json()
            for rule in rules:
                if rule["name"].startswith(rule_prefix):
                    del_resp = self.client.delete(
                        f"/projects/{self.org_slug}/{self.project_slug}/rules/{rule['id']}/"
                    )
                    del_resp.raise_for_status()
                    deleted_ids.append(rule["id"])
                    logger.info(f"Deleted legacy rule: {rule['id']} - {rule['name']}")
            return deleted_ids
        except Exception as e:
            logger.error(f"Failed to delete legacy rules: {str(e)}")
            return deleted_ids

    def create_fixed_alert(self) -> Optional[Dict[str, Any]]:
        """Create corrected alert rule with appropriate windows for payment services."""
        alert_payload = {
            "name": "Payment Service Error Alert v2",
            "conditions": [
                {
                    "id": "sentry.rules.conditions.first_seen_event.FirstSeenEventCondition",
                    "firstSeenWindow": 86400  # 24 hour aggregation window
                },
                {
                    "id": "sentry.rules.conditions.event_frequency.EventFrequencyCondition",
                    "frequencyWindow": 86400,
                    "value": 1,  # Alert on any single error
                    "comparisonType": "count"
                }
            ],
            "filters": [
                {
                    "id": "sentry.rules.filters.tagged_event.TaggedEventFilter",
                    "key": "service",
                    "value": "payment-processor"
                },
                {
                    "id": "sentry.rules.filters.issue_category.IssueCategoryFilter",
                    "value": ["error"]
                }
            ],
            "ignored_exceptions": [
                "IdempotencyKeyCollisionError",  # Known non-critical error
                "StripeWebhookDuplicateError"    # Handled by retry logic
            ],
            "actions": [
                {
                    "id": "sentry.rules.actions.notify_event.NotifyEventAction"
                },
                {
                    "id": "sentry.rules.actions.pagerduty.PagerDutyNotifyAction",
                    "account": "prod-pagerduty",
                    "service": "payment-on-call"
                }
            ],
            "actionMatch": "all",
            "frequency": 5  # Alert immediately, deduplicate for 5 minutes
        }

        try:
            resp = self.client.post(
                f"/projects/{self.org_slug}/{self.project_slug}/rules/",
                json=alert_payload
            )
            resp.raise_for_status()
            rule = resp.json()
            logger.info(f"Created fixed alert rule: {rule['id']} - {rule['name']}")
            return rule
        except Exception as e:
            logger.error(f"Failed to create fixed alert: {str(e)}")
            logger.error(f"Response: {getattr(e, 'response', {}).get('text', 'N/A')}")
            return None

if __name__ == "__main__":
    configurator = FixedSentryAlertConfigurator(
        org_slug="fintech-prod-org",
        project_slug="payment-processor-v2"
    )
    # Delete legacy misconfigured rules first
    deleted = configurator.delete_legacy_rules("Payment Service Error Alert")
    logger.info(f"Deleted {len(deleted)} legacy rules")
    # Create new fixed rule
    rule = configurator.create_fixed_alert()
    if not rule:
        sys.exit(1)
    print(json.dumps(rule, indent=2))

import os
import sys
import json
import logging
from typing import Dict, Any, List, Tuple
from sentry_api import SentryApiClient

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class SentryRuleAuditor:
    """Audits Sentry 8.0 alert rules for misconfigurations."""

    def __init__(self, org_slug: str, project_slug: str):
        self.api_token = os.getenv("SENTRY_AUTH_TOKEN")
        if not self.api_token:
            logger.error("Missing SENTRY_AUTH_TOKEN")
            sys.exit(1)
        self.client = SentryApiClient(
            base_url="https://sentry.io/api/0",
            auth_token=self.api_token
        )
        self.org_slug = org_slug
        self.project_slug = project_slug
        self.issues = []

    def _get_all_rules(self) -> List[Dict[str, Any]]:
        """Retrieve all alert rules for the project."""
        try:
            resp = self.client.get(f"/projects/{self.org_slug}/{self.project_slug}/rules/")
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            logger.error(f"Failed to fetch rules: {str(e)}")
            sys.exit(1)

    def _check_aggregation_window(self, rule: Dict[str, Any]) -> List[str]:
        """Check if aggregation window is too short for low-frequency services."""
        issues = []
        for condition in rule.get("conditions", []):
            if "firstSeenWindow" in condition:
                window = condition["firstSeenWindow"]
                if window < 3600:  # Less than 1 hour
                    issues.append(f"Aggregation window {window}s is too short for low-frequency services")
            if "frequencyWindow" in condition:
                window = condition["frequencyWindow"]
                if window < 3600:
                    issues.append(f"Frequency window {window}s is too short for low-frequency services")
        return issues

    def _check_threshold(self, rule: Dict[str, Any]) -> List[str]:
        """Check if frequency threshold is too high."""
        issues = []
        for condition in rule.get("conditions", []):
            if condition.get("id") == "sentry.rules.conditions.event_frequency.EventFrequencyCondition":
                threshold = condition.get("value", 0)
                window = condition.get("frequencyWindow", 300)
                # Calculate max possible events for the service (hardcoded for payment processor)
                max_events_per_window = (47 / 60) * window  # 47 events per minute average
                if threshold > max_events_per_window * 0.5:
                    issues.append(f"Threshold {threshold} is higher than 50% of max possible events ({max_events_per_window:.0f}) in window {window}s")
        return issues

    def _check_legacy_syntax(self, rule: Dict[str, Any]) -> List[str]:
        """Check for legacy 7.x syntax that Sentry 8.0 ignores."""
        issues = []
        legacy_keys = ["ignore_exceptions", "event_count"]
        for key in legacy_keys:
            if key in rule:
                issues.append(f"Legacy 7.x key '{key}' detected, Sentry 8.0 ignores this")
        return issues

    def audit_rules(self) -> List[Tuple[str, List[str]]]:
        """Run all audit checks against project rules."""
        rules = self._get_all_rules()
        logger.info(f"Auditing {len(rules)} alert rules")
        results = []
        for rule in rules:
            rule_issues = []
            rule_issues.extend(self._check_aggregation_window(rule))
            rule_issues.extend(self._check_threshold(rule))
            rule_issues.extend(self._check_legacy_syntax(rule))
            if rule_issues:
                results.append((rule["id"], rule["name"], rule_issues))
                logger.warning(f"Rule {rule['id']} ({rule['name']}) has {len(rule_issues)} issues")
        return results

if __name__ == "__main__":
    auditor = SentryRuleAuditor(
        org_slug="fintech-prod-org",
        project_slug="payment-processor-v2"
    )
    results = auditor.audit_rules()
    print(json.dumps([
        {
            "rule_id": r[0],
            "rule_name": r[1],
            "issues": r[2]
        } for r in results
    ], indent=2))
    if results:
        logger.error(f"Found {len(results)} misconfigured rules")
        sys.exit(1)
    else:
        logger.info("No misconfigured rules found")

name: Validate Sentry Alerts
on: [push, pull_request]
jobs:
  validate-alerts:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install sentry-api pytest
      - name: Run alert validation tests
        env:
          SENTRY_AUTH_TOKEN: ${{ secrets.SENTRY_AUTH_TOKEN }}
        run: pytest tests/test_sentry_alerts.py -v

alert_payload = {
    "conditions": [
        {
            "id": "sentry.rules.conditions.first_seen_event.FirstSeenEventCondition",
            "firstSeenWindow": 86400  # 24 hours for low-frequency services
        },
        {
            "id": "sentry.rules.conditions.event_frequency.EventFrequencyCondition",
            "frequencyWindow": 86400,
            "value": 1,  # Alert on any single error
            "comparisonType": "count"
        }
    ]
}

def _check_legacy_syntax(self, rule: Dict[str, Any]) -> List[str]:
    """Check for legacy 7.x syntax that Sentry 8.0 ignores."""
    issues = []
    legacy_keys = ["ignore_exceptions", "event_count"]
    for key in legacy_keys:
        if key in rule:
            issues.append(f"Legacy 7.x key '{key}' detected, Sentry 8.0 ignores this")
    # Check for deprecated condition IDs
    deprecated_conditions = ["sentry.rules.conditions.event_count.EventCountCondition"]
    for condition in rule.get("conditions", []):
        if condition.get("id") in deprecated_conditions:
            issues.append(f"Deprecated condition {condition['id']} detected")
    return issues