Alert Fatigue: PagerDuty 3.0 vs. Opsgenie 2.0 vs. VictorOps 8.0 for 10k Alerts/Month in 2026

import os
import time
import logging
import dataclasses
from typing import Optional, Dict, Any
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\",
    handlers=[logging.FileHandler(\"alert_ingest.log\"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

@dataclasses.dataclass
class Alert:
    \"\"\"Structured alert payload matching Prometheus metric format\"\"\"
    alert_id: str
    severity: str  # critical, warning, info
    service: str
    description: str
    metrics: Dict[str, Any]
    timestamp: float = dataclasses.field(default_factory=time.time)

class UnifiedAlertClient:
    \"\"\"Client to ingest alerts to PagerDuty 3.0, Opsgenie 2.0, and VictorOps 8.0\"\"\"

    def __init__(self):
        # Load API keys from environment variables (never hardcode!)
        self.pagerduty_key = os.getenv(\"PAGERDUTY_API_KEY\")
        self.opsgenie_key = os.getenv(\"OPSGENIE_API_KEY\")
        self.victorops_key = os.getenv(\"VICTOROPS_API_KEY\")

        # Configure retry strategy for transient network errors
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=[\"POST\"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session = requests.Session()
        self.session.mount(\"https://\", adapter)
        self.session.mount(\"http://\", adapter)

        # Validate all API keys are present
        if not all([self.pagerduty_key, self.opsgenie_key, self.victorops_key]):
            raise ValueError(\"Missing one or more API keys. Set PAGERDUTY_API_KEY, OPSGENIE_API_KEY, VICTOROPS_API_KEY\")

    def _send_pagerduty(self, alert: Alert) -> bool:
        \"\"\"Send alert to PagerDuty 3.0 Events API v2\"\"\"
        url = \"https://events.pagerduty.com/v2/enqueue\"
        payload = {
            \"routing_key\": self.pagerduty_key,
            \"event_action\": \"trigger\",
            \"payload\": {
                \"summary\": alert.description,
                \"severity\": alert.severity,
                \"source\": alert.service,
                \"timestamp\": time.strftime(\"%Y-%m-%dT%H:%M:%SZ\", time.gmtime(alert.timestamp)),
                \"custom_details\": alert.metrics
            },
            \"dedup_key\": alert.alert_id
        }
        try:
            resp = self.session.post(url, json=payload, timeout=10)
            resp.raise_for_status()
            logger.info(f\"PagerDuty: Successfully ingested alert {alert.alert_id}\")
            return True
        except requests.exceptions.RequestException as e:
            logger.error(f\"PagerDuty: Failed to ingest alert {alert.alert_id}: {str(e)}\")
            return False

    def _send_opsgenie(self, alert: Alert) -> bool:
        \"\"\"Send alert to Opsgenie 2.0 Alert API v2\"\"\"
        url = \"https://api.opsgenie.com/v2/alerts\"
        headers = {
            \"Authorization\": f\"GenieKey {self.opsgenie_key}\",
            \"Content-Type\": \"application/json\"
        }
        payload = {
            \"message\": alert.description,
            \"alias\": alert.alert_id,
            \"severity\": alert.severity.upper(),
            \"entity\": alert.service,
            \"details\": alert.metrics,
            \"createdAt\": time.strftime(\"%Y-%m-%dT%H:%M:%SZ\", time.gmtime(alert.timestamp))
        }
        try:
            resp = self.session.post(url, json=payload, headers=headers, timeout=10)
            resp.raise_for_status()
            logger.info(f\"Opsgenie: Successfully ingested alert {alert.alert_id}\")
            return True
        except requests.exceptions.RequestException as e:
            logger.error(f\"Opsgenie: Failed to ingest alert {alert.alert_id}: {str(e)}\")
            return False

    def _send_victorops(self, alert: Alert) -> bool:
        \"\"\"Send alert to VictorOps 8.0 REST API v1\"\"\"
        url = \"https://api.victorops.com/api-public/v1/alerts\"
        headers = {
            \"X-VO-Api-Id\": self.victorops_key.split(\":\")[0],
            \"X-VO-Api-Key\": self.victorops_key.split(\":\")[1],
            \"Content-Type\": \"application/json\"
        }
        payload = {
            \"message\": alert.description,
            \"entity_id\": alert.alert_id,
            \"state\": \"CRITICAL\" if alert.severity == \"critical\" else \"WARNING\",
            \"service\": alert.service,
            \"timestamp\": int(alert.timestamp),
            \"details\": alert.metrics
        }
        try:
            resp = self.session.post(url, json=payload, headers=headers, timeout=10)
            resp.raise_for_status()
            logger.info(f\"VictorOps: Successfully ingested alert {alert.alert_id}\")
            return True
        except requests.exceptions.RequestException as e:
            logger.error(f\"VictorOps: Failed to ingest alert {alert.alert_id}: {str(e)}\")
            return False

    def send_alert(self, alert: Alert) -> Dict[str, bool]:
        \"\"\"Send alert to all three tools, return per-tool success status\"\"\"
        return {
            \"pagerduty\": self._send_pagerduty(alert),
            \"opsgenie\": self._send_opsgenie(alert),
            \"victorops\": self._send_victorops(alert)
        }

if __name__ == \"__main__\":
    # Example usage: Send a test critical alert
    try:
        client = UnifiedAlertClient()
        test_alert = Alert(
            alert_id=\"alert-2026-05-01-001\",
            severity=\"critical\",
            service=\"payment-gateway\",
            description=\"Payment gateway p99 latency exceeds 2s threshold\",
            metrics={\"p99_latency_ms\": 2100, \"error_rate\": 0.05, \"requests_per_sec\": 450}
        )
        results = client.send_alert(test_alert)
        logger.info(f\"Alert ingestion results: {results}\")
    except Exception as e:
        logger.critical(f\"Failed to initialize alert client: {str(e)}\")
        exit(1)

import os
import time
import logging
from typing import List, Dict, Optional
import requests
from datetime import datetime, timedelta

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

class OpsgenieDeduplicator:
    \"\"\"Leverage Opsgenie 2.0 ML-based deduplication to reduce false positives\"\"\"

    def __init__(self):
        self.api_key = os.getenv(\"OPSGENIE_API_KEY\")
        if not self.api_key:
            raise ValueError(\"OPSGENIE_API_KEY environment variable not set\")
        self.base_url = \"https://api.opsgenie.com/v2\"
        self.headers = {
            \"Authorization\": f\"GenieKey {self.api_key}\",
            \"Content-Type\": \"application/json\"
        }
        self.session = requests.Session()
        self.session.headers.update(self.headers)

    def fetch_recent_alerts(self, hours: int = 24) -> List[Dict]:
        \"\"\"Fetch all alerts from the last N hours\"\"\"
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(hours=hours)
        params = {
            \"query\": \"status:open\",
            \"createdAt>\": start_time.strftime(\"%Y-%m-%dT%H:%M:%SZ\"),
            \"createdAt<\": end_time.strftime(\"%Y-%m-%dT%H:%M:%SZ\"),
            \"limit\": 1000  # Max limit per Opsgenie API
        }
        alerts = []
        next_url = f\"{self.base_url}/alerts\"
        while next_url:
            try:
                resp = self.session.get(next_url, params=params, timeout=15)
                resp.raise_for_status()
                data = resp.json()
                alerts.extend(data.get(\"data\", []))
                next_url = data.get(\"paging\", {}).get(\"next\")
                params = {}  # Clear params for paginated requests
            except requests.exceptions.RequestException as e:
                logger.error(f\"Failed to fetch alerts: {str(e)}\")
                break
        logger.info(f\"Fetched {len(alerts)} open alerts from last {hours} hours\")
        return alerts

    def get_deduplication_groups(self) -> List[Dict]:
        \"\"\"Retrieve ML-generated deduplication groups from Opsgenie 2.0\"\"\"
        url = f\"{self.base_url}/alerts/deduplication-groups\"
        try:
            resp = self.session.get(url, timeout=15)
            resp.raise_for_status()
            groups = resp.json().get(\"data\", [])
            logger.info(f\"Found {len(groups)} deduplication groups\")
            return groups
        except requests.exceptions.RequestException as e:
            logger.error(f\"Failed to fetch deduplication groups: {str(e)}\")
            return []

    def flag_false_positives(self, group: Dict) -> bool:
        \"\"\"
        Flag deduplication groups as false positives if:
        1. All alerts in group have severity < critical
        2. Group has no associated incident
        3. Alert count > 5 (noise threshold)
        \"\"\"
        alert_ids = group.get(\"alertIds\", [])
        if len(alert_ids) < 5:
            return False
        # Fetch full alert details for the group (batch request)
        batch_url = f\"{self.base_url}/alerts/batch-get\"
        payload = {\"alertIds\": alert_ids[:100]}  # Max 100 per batch
        try:
            resp = self.session.post(batch_url, json=payload, timeout=15)
            resp.raise_for_status()
            alerts = resp.json().get(\"data\", [])
            # Check if any alert is critical
            has_critical = any(a.get(\"severity\") == \"CRITICAL\" for a in alerts)
            # Check if group has associated incident
            has_incident = group.get(\"incidentId\") is not None
            if not has_critical and not has_incident:
                logger.info(f\"Flagging group {group.get('id')} as false positive (count: {len(alert_ids)})\")
                return True
            return False
        except requests.exceptions.RequestException as e:
            logger.error(f\"Failed to check false positive for group {group.get('id')}: {str(e)}\")
            return False

    def suppress_false_positives(self, dry_run: bool = True) -> int:
        \"\"\"Suppress all flagged false positive alerts, return count suppressed\"\"\"
        groups = self.get_deduplication_groups()
        suppressed = 0
        for group in groups:
            if self.flag_false_positives(group):
                alert_ids = group.get(\"alertIds\", [])
                for alert_id in alert_ids:
                    if dry_run:
                        logger.info(f\"DRY RUN: Would suppress alert {alert_id}\")
                        suppressed +=1
                    else:
                        suppress_url = f\"{self.base_url}/alerts/{alert_id}/suppress\"
                        try:
                            resp = self.session.post(suppress_url, timeout=10)
                            resp.raise_for_status()
                            suppressed +=1
                            logger.info(f\"Suppressed false positive alert {alert_id}\")
                        except requests.exceptions.RequestException as e:
                            logger.error(f\"Failed to suppress alert {alert_id}: {str(e)}\")
        logger.info(f\"Total false positives {'would be ' if dry_run else ''}suppressed: {suppressed}\")
        return suppressed

if __name__ == \"__main__\":
    try:
        deduplicator = OpsgenieDeduplicator()
        # Run dry run first to validate
        print(\"Running dry run for false positive suppression...\")
        deduplicator.suppress_false_positives(dry_run=True)
        # Uncomment to apply suppression
        # deduplicator.suppress_false_positives(dry_run=False)
    except Exception as e:
        logger.critical(f\"Deduplicator failed: {str(e)}\")
        exit(1)

import os
import time
import logging
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import requests

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

class VictorOpsEscalationManager:
    \"\"\"Automate on-call escalation and shift management for VictorOps 8.0\"\"\"

    def __init__(self):
        # VictorOps API uses API ID and Key separated by colon
        vo_creds = os.getenv(\"VICTOROPS_API_CREDENTIALS\")
        if not vo_creds or \":\" not in vo_creds:
            raise ValueError(\"Set VICTOROPS_API_CREDENTIALS to 'api_id:api_key'\")
        self.api_id, self.api_key = vo_creds.split(\":\", 1)
        self.base_url = \"https://api.victorops.com/api-public/v1\"
        self.headers = {
            \"X-VO-Api-Id\": self.api_id,
            \"X-VO-Api-Key\": self.api_key,
            \"Content-Type\": \"application/json\"
        }
        self.session = requests.Session()
        self.session.headers.update(self.headers)

    def get_oncall_team(self, team_id: str) -> List[Dict]:
        \"\"\"Retrieve current on-call engineers for a team\"\"\"
        url = f\"{self.base_url}/teams/{team_id}/oncalls\"
        try:
            resp = self.session.get(url, timeout=15)
            resp.raise_for_status()
            oncalls = resp.json().get(\"oncalls\", [])
            logger.info(f\"Team {team_id} has {len(oncalls)} on-call engineers\")
            return oncalls
        except requests.exceptions.RequestException as e:
            logger.error(f\"Failed to fetch on-call team {team_id}: {str(e)}\")
            return []

    def trigger_escalation(self, alert_id: str, team_id: str, escalation_policy_id: str) -> bool:
        \"\"\"Trigger escalation for an unresolved alert after 5 minutes\"\"\"
        # Check if alert is still open
        alert_url = f\"{self.base_url}/alerts/{alert_id}\"
        try:
            resp = self.session.get(alert_url, timeout=10)
            resp.raise_for_status()
            alert = resp.json()
            if alert.get(\"state\") not in [\"CRITICAL\", \"WARNING\"]:
                logger.info(f\"Alert {alert_id} is already resolved, skipping escalation\")
                return False
        except requests.exceptions.RequestException as e:
            logger.error(f\"Failed to check alert {alert_id} status: {str(e)}\")
            return False

        # Trigger escalation via VictorOps API
        esc_url = f\"{self.base_url}/teams/{team_id}/escalation-policies/{escalation_policy_id}/execute\"
        payload = {
            \"alertId\": alert_id,
            \"message\": f\"Escalating unresolved alert {alert_id} after 5 minute SLA breach\"
        }
        try:
            resp = self.session.post(esc_url, json=payload, timeout=15)
            resp.raise_for_status()
            logger.info(f\"Successfully triggered escalation for alert {alert_id}\")
            return True
        except requests.exceptions.RequestException as e:
            logger.error(f\"Failed to trigger escalation for alert {alert_id}: {str(e)}\")
            return False

    def rotate_shifts(self, team_id: str, rotation_id: str) -> bool:
        \"\"\"Rotate on-call shift for a team (runs daily at 9am UTC)\"\"\"
        url = f\"{self.base_url}/teams/{team_id}/rotations/{rotation_id}/rotate\"
        payload = {
            \"rotationTime\": datetime.utcnow().strftime(\"%Y-%m-%dT%H:%M:%SZ\")
        }
        try:
            resp = self.session.post(url, json=payload, timeout=15)
            resp.raise_for_status()
            logger.info(f\"Successfully rotated shift for rotation {rotation_id}\")
            return True
        except requests.exceptions.RequestException as e:
            logger.error(f\"Failed to rotate shift {rotation_id}: {str(e)}\")
            return False

    def get_escalation_metrics(self, days: int = 7) -> Dict:
        \"\"\"Retrieve escalation time metrics for the last N days\"\"\"
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(days=days)
        url = f\"{self.base_url}/metrics/escalations\"
        params = {
            \"start\": start_time.strftime(\"%Y-%m-%dT%H:%M:%SZ\"),
            \"end\": end_time.strftime(\"%Y-%m-%dT%H:%M:%SZ\"),
            \"teamId\": \"all\"
        }
        try:
            resp = self.session.get(url, params=params, timeout=15)
            resp.raise_for_status()
            metrics = resp.json().get(\"data\", {})
            avg_esc_time = metrics.get(\"averageEscalationTimeMinutes\", 0)
            total_escalations = metrics.get(\"totalEscalations\", 0)
            logger.info(f\"7-day avg escalation time: {avg_esc_time}min, total escalations: {total_escalations}\")
            return metrics
        except requests.exceptions.RequestException as e:
            logger.error(f\"Failed to fetch escalation metrics: {str(e)}\")
            return {}

if __name__ == \"__main__\":
    try:
        manager = VictorOpsEscalationManager()
        # Example: Get on-call team and trigger escalation for test alert
        team_id = \"team-payment-001\"
        oncalls = manager.get_oncall_team(team_id)
        if oncalls:
            print(f\"Current on-call engineers: {[o.get('user', {}).get('username') for o in oncalls]}\")
            # Simulate triggering escalation for test alert
            test_alert_id = \"victorops-alert-2026-05-01-001\"
            esc_policy_id = \"esc-policy-payment-001\"
            manager.trigger_escalation(test_alert_id, team_id, esc_policy_id)
        # Get weekly escalation metrics
        metrics = manager.get_escalation_metrics(days=7)
    except Exception as e:
        logger.critical(f\"Escalation manager failed: {str(e)}\")
        exit(1)

# Short snippet: Initialize unified client
from unified_alert_client import UnifiedAlertClient
import os

os.environ[\"PAGERDUTY_API_KEY\"] = \"your-pd-key\"
os.environ[\"OPSGENIE_API_KEY\"] = \"your-og-key\"
os.environ[\"VICTOROPS_API_KEY\"] = \"your-vo-key\"

client = UnifiedAlertClient()
# Route 1% of alerts to Opsgenie for A/B test
if hash(alert.alert_id) % 100 == 0:
    client.send_alert(alert, tools=[\"opsgenie\"])
else:
    client.send_alert(alert, tools=[\"pagerduty\", \"victorops\"])

# Short snippet: Check Opsgenie deduplication groups
groups = deduplicator.get_deduplication_groups()
for group in groups:
    if len(group[\"alertIds\"]) > 5:
        print(f\"Group {group['id']} has {len(group['alertIds'])} alerts\")

# Short snippet: Terraform for VictorOps on-call rotation
resource \"victorops_rotation\" \"payment_oncall\" {
  team_id = \"team-payment-001\"
  name    = \"Payment Gateway On-Call\"
  type    = \"daily\"
  start_time = \"2026-05-01T09:00:00Z\"
  members = [\"user-001\", \"user-002\", \"user-003\"]
}