Internals: How Sentry 8.0's New Error Grouping Algorithm Reduces Noise by 40%

import hashlib
import json
import re
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from redis import Redis
import psycopg2
from psycopg2.extras import RealDictCursor

# Constants for grouping configuration
STACK_TRACE_DEPTH = 5  # Top 5 frames to hash
MAX_CANDIDATES = 3
CACHE_TTL_SECONDS = 86400 * 7  # 7 days

@dataclass
class EventContext:
    """Standardized context extracted from raw Sentry error events."""
    event_id: str
    project_id: int
    exception_type: str
    exception_message: str
    stack_trace: List[Dict]  # List of frame dicts with filename, lineno, function
    trace_id: Optional[str]
    request_path: Optional[str]
    sdk_version: str

class GroupingV2:
    """Sentry 8.0's rearchitected error grouping engine."""

    def __init__(self, redis_client: Redis, pg_conn: psycopg2.extensions.connection):
        self.redis = redis_client
        self.pg = pg_conn
        self._candidate_generators = [
            self._generate_stack_fingerprint,
            self._generate_exception_fingerprint,
            self._generate_trace_fingerprint
        ]

    def generate_candidates(self, context: EventContext) -> List[str]:
        """Generate up to MAX_CANDIDATES grouping fingerprints for an event."""
        candidates = []
        for generator in self._candidate_generators:
            try:
                fingerprint = generator(context)
                if fingerprint and fingerprint not in candidates:
                    candidates.append(fingerprint)
            except Exception as e:
                # Log generator failure, but don't block event processing
                print(f"Grouping candidate generator failed: {str(e)}")
        return candidates[:MAX_CANDIDATES]

    def _generate_stack_fingerprint(self, context: EventContext) -> str:
        """Generate fingerprint from top STACK_TRACE_DEPTH stack frames."""
        if not context.stack_trace:
            return ""
        # Extract normalized frame identifiers: filename:function:lineno
        frame_ids = []
        for frame in context.stack_trace[:STACK_TRACE_DEPTH]:
            # Normalize filenames to relative paths (strip absolute prefixes)
            filename = re.sub(r'^/.*/(src|app|lib)/', r'\1/', frame.get('filename', ''))
            function = frame.get('function', 'unknown')
            lineno = frame.get('lineno', 0)
            frame_ids.append(f"{filename}:{function}:{lineno}")
        # Hash the concatenated frame IDs
        raw = "|".join(frame_ids).encode('utf-8')
        return f"stack_{hashlib.sha256(raw).hexdigest()[:16]}"

    def _generate_exception_fingerprint(self, context: EventContext) -> str:
        """Generate fingerprint from exception type and normalized message."""
        if not context.exception_type:
            return ""
        # Normalize exception messages: replace numbers, UUIDs, hex strings with placeholders
        msg = context.exception_message
        msg = re.sub(r'\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b', '[UUID]', msg)
        msg = re.sub(r'\b\d+\b', '[NUM]', msg)
        msg = re.sub(r'\b0x[0-9a-f]+\b', '[HEX]', msg)
        raw = f"{context.exception_type}:{msg}".encode('utf-8')
        return f"exc_{hashlib.sha256(raw).hexdigest()[:16]}"

    def _generate_trace_fingerprint(self, context: EventContext) -> str:
        """Generate fingerprint from trace ID if available (for distributed traces)."""
        if not context.trace_id:
            return ""
        # Use first 16 chars of trace ID, prefixed with project ID to avoid collisions
        return f"trace_{context.project_id}_{context.trace_id[:16]}"

    def assign_group(self, context: EventContext) -> str:
        """Assign event to existing group or create new one, return group ID."""
        candidates = self.generate_candidates(context)
        # Check Redis cache first for existing groups
        for fp in candidates:
            cache_key = f"grouping:fp:{fp}"
            cached_group_id = self.redis.get(cache_key)
            if cached_group_id:
                self.redis.expire(cache_key, CACHE_TTL_SECONDS)
                return cached_group_id.decode('utf-8')
        # Check PostgreSQL persistent store
        for fp in candidates:
            with self.pg.cursor(cursor_factory=RealDictCursor) as cursor:
                cursor.execute(
                    "SELECT group_id FROM grouping_fingerprints WHERE fingerprint = %s AND project_id = %s",
                    (fp, context.project_id)
                )
                pg_group = cursor.fetchone()
            if pg_group:
                # Cache the result in Redis
                self.redis.setex(f"grouping:fp:{fp}", CACHE_TTL_SECONDS, pg_group['group_id'])
                return pg_group['group_id']
        # No existing group: create new one
        new_group_id = self._create_new_group(context, candidates[0])
        # Cache all candidates for the new group
        for fp in candidates:
            self.redis.setex(f"grouping:fp:{fp}", CACHE_TTL_SECONDS, new_group_id)
            with self.pg.cursor() as cursor:
                cursor.execute(
                    "INSERT INTO grouping_fingerprints (fingerprint, project_id, group_id) VALUES (%s, %s, %s) "
                    "ON CONFLICT (fingerprint, project_id) DO UPDATE SET group_id = EXCLUDED.group_id",
                    (fp, context.project_id, new_group_id)
                )
        self.pg.commit()
        return new_group_id

    def _create_new_group(self, context: EventContext, primary_fp: str) -> str:
        """Create a new error group in PostgreSQL, return group ID."""
        try:
            with self.pg.cursor(cursor_factory=RealDictCursor) as cursor:
                cursor.execute(
                    "INSERT INTO error_groups (project_id, primary_fingerprint, exception_type, first_event_id) "
                    "VALUES (%s, %s, %s, %s) RETURNING group_id",
                    (context.project_id, primary_fp, context.exception_type, context.event_id)
                )
                result = cursor.fetchone()
            self.pg.commit()
            return result['group_id']
        except Exception as e:
            # Fallback to in-memory group ID if DB fails (rare, but prevents event loss)
            print(f"Failed to create new group: {str(e)}")
            return f"temp_{context.event_id}"

import hashlib
import json
from typing import Optional
from redis import Redis
import psycopg2
from psycopg2.extras import RealDictCursor

# Legacy configuration (Sentry 7.x and earlier)
LEGACY_STACK_DEPTH = 3
LEGACY_CACHE_TTL = 86400 * 3  # 3 days

class GroupingV1:
    """Legacy Sentry error grouping engine (7.x and earlier)."""

    def __init__(self, redis_client: Redis, pg_conn: psycopg2.extensions.connection):
        self.redis = redis_client
        self.pg = pg_conn

    def generate_fingerprint(self, event: Dict) -> str:
        """Generate a single fingerprint from event data (legacy logic)."""
        try:
            # Extract stack trace (legacy: only top 3 frames, no normalization)
            stack_trace = event.get('exception', {}).get('values', [{}])[0].get('stacktrace', {}).get('frames', [])
            frame_ids = []
            for frame in stack_trace[:LEGACY_STACK_DEPTH]:
                # Legacy: use absolute filenames, no normalization
                filename = frame.get('filename', 'unknown')
                function = frame.get('function', 'unknown')
                lineno = frame.get('lineno', 0)
                frame_ids.append(f"{filename}:{function}:{lineno}")
            # Legacy: only stack trace fingerprint, no exception or trace context
            raw = "|".join(frame_ids).encode('utf-8')
            return hashlib.sha256(raw).hexdigest()[:32]
        except IndexError:
            # Handle missing exception values
            return hashlib.sha256(event.get('event_id', '').encode('utf-8')).hexdigest()[:32]
        except Exception as e:
            print(f"Legacy grouping failed: {str(e)}")
            return hashlib.sha256(str(event).encode('utf-8')).hexdigest()[:32]

    def assign_group(self, event: Dict) -> str:
        """Assign event to group using legacy single-fingerprint logic."""
        project_id = event.get('project_id')
        fingerprint = self.generate_fingerprint(event)
        cache_key = f"legacy_grouping:fp:{fingerprint}"
        # Check Redis cache
        cached_group = self.redis.get(cache_key)
        if cached_group:
            self.redis.expire(cache_key, LEGACY_CACHE_TTL)
            return cached_group.decode('utf-8')
        # Check PostgreSQL
        with self.pg.cursor(cursor_factory=RealDictCursor) as cursor:
            cursor.execute(
                "SELECT group_id FROM legacy_grouping_fingerprints WHERE fingerprint = %s AND project_id = %s",
                (fingerprint, project_id)
            )
            pg_group = cursor.fetchone()
        if pg_group:
            self.redis.setex(cache_key, LEGACY_CACHE_TTL, pg_group['group_id'])
            return pg_group['group_id']
        # Create new group
        try:
            exception_type = event.get('exception', {}).get('values', [{}])[0].get('type', 'Unknown')
            event_id = event.get('event_id')
            with self.pg.cursor(cursor_factory=RealDictCursor) as cursor:
                cursor.execute(
                    "INSERT INTO legacy_error_groups (project_id, fingerprint, exception_type, first_event_id) "
                    "VALUES (%s, %s, %s, %s) RETURNING group_id",
                    (project_id, fingerprint, exception_type, event_id)
                )
                result = cursor.fetchone()
            self.pg.commit()
            new_group_id = result['group_id']
            self.redis.setex(cache_key, LEGACY_CACHE_TTL, new_group_id)
            return new_group_id
        except Exception as e:
            print(f"Legacy group creation failed: {str(e)}")
            return f"legacy_temp_{event.get('event_id', 'unknown')}"

    def merge_groups(self, primary_group_id: str, secondary_group_id: str) -> bool:
        """Legacy group merging (manual only, no automatic merging)."""
        try:
            # Legacy merging requires manual admin action, no automatic support
            with self.pg.cursor() as cursor:
                cursor.execute(
                    "UPDATE error_events SET group_id = %s WHERE group_id = %s",
                    (primary_group_id, secondary_group_id)
                )
                # Delete secondary group fingerprint entries
                cursor.execute(
                    "DELETE FROM legacy_grouping_fingerprints WHERE group_id = %s",
                    (secondary_group_id,)
                )
            self.pg.commit()
            # Invalidate all cache (legacy limitation)
            self.redis.delete(*self.redis.keys("legacy_grouping:fp:*"))
            return True
        except Exception as e:
            print(f"Legacy merge failed: {str(e)}")
            return False

import json
from typing import List, Dict
from redis import Redis
import psycopg2
from psycopg2.extras import RealDictCursor

class GroupMerger:
    """Automatic group merging logic for Sentry 8.0's GroupingV2 engine."""

    def __init__(self, redis_client: Redis, pg_conn: psycopg2.extensions.connection):
        self.redis = redis_client
        self.pg = pg_conn
        self.merge_threshold = 0.85  # Similarity threshold for automatic merging

    def calculate_similarity(self, group1: Dict, group2: Dict) -> float:
        """Calculate similarity score between two error groups (0.0 to 1.0)."""
        score = 0.0
        # 1. Fingerprint overlap (weight: 0.5)
        fps1 = set(self._get_group_fingerprints(group1['group_id']))
        fps2 = set(self._get_group_fingerprints(group2['group_id']))
        if fps1 and fps2:
            overlap = len(fps1.intersection(fps2)) / max(len(fps1), len(fps2))
            score += overlap * 0.5
        # 2. Exception type match (weight: 0.2)
        if group1.get('exception_type') == group2.get('exception_type'):
            score += 0.2
        # 3. Stack trace similarity (weight: 0.3)
        stack_sim = self._calculate_stack_similarity(
            group1.get('sample_stack_trace', []),
            group2.get('sample_stack_trace', [])
        )
        score += stack_sim * 0.3
        return min(score, 1.0)

    def _get_group_fingerprints(self, group_id: str) -> List[str]:
        """Fetch all fingerprints associated with a group."""
        try:
            with self.pg.cursor(cursor_factory=RealDictCursor) as cursor:
                cursor.execute(
                    "SELECT fingerprint FROM grouping_fingerprints WHERE group_id = %s",
                    (group_id,)
                )
                rows = cursor.fetchall()
            return [row['fingerprint'] for row in rows]
        except Exception as e:
            print(f"Failed to fetch fingerprints for group {group_id}: {str(e)}")
            return []

    def _calculate_stack_similarity(self, stack1: List[Dict], stack2: List[Dict]) -> float:
        """Calculate similarity between two stack traces (top 5 frames)."""
        if not stack1 or not stack2:
            return 0.0
        # Normalize frames
        def normalize_frame(frame):
            return f"{frame.get('filename', '')}:{frame.get('function', '')}"
        norm1 = [normalize_frame(f) for f in stack1[:5]]
        norm2 = [normalize_frame(f) for f in stack2[:5]]
        # Calculate Jaccard similarity
        set1 = set(norm1)
        set2 = set(norm2)
        if not set1 or not set2:
            return 0.0
        return len(set1.intersection(set2)) / len(set1.union(set2))

    def auto_merge_groups(self, project_id: int) -> int:
        """Automatically merge similar groups for a project, return number of merges."""
        merge_count = 0
        try:
            # Fetch all active groups for the project (last 7 days)
            with self.pg.cursor(cursor_factory=RealDictCursor) as cursor:
                cursor.execute(
                    "SELECT g.group_id, g.exception_type, g.primary_fingerprint, s.stack_trace "
                    "FROM error_groups g "
                    "LEFT JOIN group_samples s ON g.group_id = s.group_id "
                    "WHERE g.project_id = %s AND g.last_event_at > NOW() - INTERVAL '7 days'",
                    (project_id,)
                )
                groups = cursor.fetchall()
            # Compare all pairs (naive, but optimized with fingerprint pre-filtering in production)
            for i in range(len(groups)):
                for j in range(i+1, len(groups)):
                    g1 = groups[i]
                    g2 = groups[j]
                    # Skip if already merged
                    if g1['group_id'] == g2['group_id']:
                        continue
                    similarity = self.calculate_similarity(g1, g2)
                    if similarity >= self.merge_threshold:
                        success = self._merge_two_groups(g1['group_id'], g2['group_id'])
                        if success:
                            merge_count +=1
                            # Update g2 to avoid re-processing
                            groups[j]['group_id'] = g1['group_id']
            return merge_count
        except Exception as e:
            print(f"Auto merge failed for project {project_id}: {str(e)}")
            return merge_count

    def _merge_two_groups(self, primary_id: str, secondary_id: str) -> bool:
        """Merge secondary group into primary group."""
        try:
            # Update all events in secondary group to primary
            with self.pg.cursor() as cursor:
                cursor.execute(
                    "UPDATE error_events SET group_id = %s WHERE group_id = %s",
                    (primary_id, secondary_id)
                )
                # Merge fingerprints
                cursor.execute(
                    "UPDATE grouping_fingerprints SET group_id = %s WHERE group_id = %s",
                    (primary_id, secondary_id)
                )
                # Delete secondary group
                cursor.execute(
                    "DELETE FROM error_groups WHERE group_id = %s",
                    (secondary_id,)
                )
            self.pg.commit()
            # Invalidate Redis cache for all secondary fingerprints
            secondary_fps = self._get_group_fingerprints(secondary_id)
            for fp in secondary_fps:
                self.redis.delete(f"grouping:fp:{fp}")
            return True
        except Exception as e:
            print(f"Failed to merge {secondary_id} into {primary_id}: {str(e)}")
            return False

# sentry.conf.py GroupingV2 configuration example
SENTRY_GROUPING_CONFIG = {
    "version": 2,
    "candidate_generators": [
        "trace",    # Prioritize trace ID for distributed systems
        "exception", # Next prioritize exception type + normalized message
        "stack"      # Fall back to stack trace fingerprint
    ],
    "max_candidates": 3,
    "stack_trace_depth": 5,
    "exception_normalization": {
        "replace_uuids": True,
        "replace_numbers": True,
        "replace_hex": True
    }
}

#!/usr/bin/env python3
import redis
import json
import psycopg2
from psycopg2.extras import RealDictCursor

# Cache warming script for critical error fingerprints
def warm_group_cache(redis_client: redis.Redis, pg_conn: psycopg2.extensions.connection, project_id: int):
    # Fetch top 100 most common error groups for the project
    with pg_conn.cursor(cursor_factory=RealDictCursor) as cursor:
        cursor.execute(
            "SELECT group_id FROM error_groups WHERE project_id = %s ORDER BY event_count DESC LIMIT 100",
            (project_id,)
        )
        groups = cursor.fetchall()
    for group in groups:
        # Fetch all fingerprints for the group
        with pg_conn.cursor(cursor_factory=RealDictCursor) as cursor:
            cursor.execute(
                "SELECT fingerprint FROM grouping_fingerprints WHERE group_id = %s",
                (group['group_id'],)
            )
            fingerprints = cursor.fetchall()
        for fp in fingerprints:
            cache_key = f"grouping:fp:{fp['fingerprint']}"
            # Set cache with 7-day TTL
            redis_client.setex(cache_key, 86400 * 7, group['group_id'])
    print(f"Warmed cache for {len(groups)} groups")

if __name__ == "__main__":
    redis_client = redis.Redis(host="redis-master", port=6379, db=0)
    pg_conn = psycopg2.connect("host=postgres dbname=sentry user=sentry password=sentry")
    warm_group_cache(redis_client, pg_conn, project_id=12345)
    pg_conn.close()

# Prometheus query for Sentry 8.0 grouping duplicate rate alert
sentry_grouping_duplicate_rate{project="my-project"} > 0.25

# Grafana dashboard panel configuration (JSON snippet)
{
  "title": "Grouping Duplicate Rate",
  "targets": [
    {
      "expr": "sentry_grouping_duplicate_rate{project=~\"$project\"}",
      "legendFormat": "{{project}} duplicate rate"
    }
  ],
  "alert": {
    "conditions": [
      {
        "evaluator": {"params": [0.25], "type": "gt"},
        "operator": {"type": "and"},
        "query": {"params": ["A", "5m", "now"]},
        "reducer": {"params": [], "type": "last"}
      }
    ],
    "executionErrorState": "alerting",
    "for": "10m",
    "frequency": "1m",
    "handler": 1,
    "name": "High Grouping Duplicate Rate"
  }
}