Skip to content

chore(aci): manually add spans in delayed processing task #91764

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
May 15, 2025
Merged

chore(aci): manually add spans in delayed processing task #91764

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
110 changes: 67 additions & 43 deletions src/sentry/rules/processing/delayed_processing.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from datetime import datetime, timedelta, timezone
from typing import Any, DefaultDict, NamedTuple

import sentry_sdk
from celery import Task
from django.db.models import OuterRef, Subquery

Expand Down Expand Up @@ -584,24 +585,33 @@ def apply_delayed(project_id: int, batch_key: str | None = None, *args: Any, **k
"""
Grab rules, groups, and events from the Redis buffer, evaluate the "slow" conditions in a bulk snuba query, and fire them if they pass
"""
project = fetch_project(project_id)
if not project:
return

rulegroup_to_event_data = fetch_rulegroup_to_event_data(project_id, batch_key)
rules_to_groups = get_rules_to_groups(rulegroup_to_event_data)
alert_rules = fetch_alert_rules(list(rules_to_groups.keys()))
condition_groups = get_condition_query_groups(alert_rules, rules_to_groups)
logger.info(
"delayed_processing.condition_groups",
extra={
"condition_groups": len(condition_groups),
"project_id": project_id,
"rules_to_groups": rules_to_groups,
},
)
with sentry_sdk.start_span(
op="delayed_processing.prepare_data", name="Fetch data from buffers in delayed processing"
):
project = fetch_project(project_id)
if not project:
return

rulegroup_to_event_data = fetch_rulegroup_to_event_data(project_id, batch_key)
rules_to_groups = get_rules_to_groups(rulegroup_to_event_data)
alert_rules = fetch_alert_rules(list(rules_to_groups.keys()))
condition_groups = get_condition_query_groups(alert_rules, rules_to_groups)
logger.info(
"delayed_processing.condition_groups",
extra={
"condition_groups": len(condition_groups),
"project_id": project_id,
"rules_to_groups": rules_to_groups,
},
)

with metrics.timer("delayed_processing.get_condition_group_results.duration"):
with (
metrics.timer("delayed_processing.get_condition_group_results.duration"),
sentry_sdk.start_span(
op="delayed_processing.get_condition_group_results",
name="Fetch condition group results in delayed processing",
),
):
condition_group_results = get_condition_group_results(condition_groups, project)

has_workflow_engine = features.has(
Expand All @@ -625,34 +635,48 @@ def apply_delayed(project_id: int, batch_key: str | None = None, *args: Any, **k
for rule in alert_rules:
rules_to_slow_conditions[rule].extend(get_slow_conditions(rule))

rules_to_fire = defaultdict(set)
if condition_group_results:
rules_to_fire = get_rules_to_fire(
condition_group_results, rules_to_slow_conditions, rules_to_groups, project.id
)
if has_workflow_engine or features.has("projects:num-events-issue-debugging", project):
logger.info(
"delayed_processing.rules_to_fire",
extra={
"rules_to_fire": {rule.id: groups for rule, groups in rules_to_fire.items()},
"project_id": project_id,
"rules_to_slow_conditions": {
rule.id: conditions for rule, conditions in rules_to_slow_conditions.items()
},
"rules_to_groups": rules_to_groups,
},
)
if random.random() < 0.01:
logger.info(
"delayed_processing.rule_to_fire",
extra={"rules_to_fire": list(rules_to_fire.keys()), "project_id": project_id},
with sentry_sdk.start_span(
op="delayed_processing.get_rules_to_fire",
name="Process rule conditions in delayed processing",
):
rules_to_fire = defaultdict(set)
if condition_group_results:
rules_to_fire = get_rules_to_fire(
condition_group_results, rules_to_slow_conditions, rules_to_groups, project.id
)
if has_workflow_engine or features.has("projects:num-events-issue-debugging", project):
logger.info(
"delayed_processing.rules_to_fire",
extra={
"rules_to_fire": {
rule.id: groups for rule, groups in rules_to_fire.items()
},
"project_id": project_id,
"rules_to_slow_conditions": {
rule.id: conditions
for rule, conditions in rules_to_slow_conditions.items()
},
"rules_to_groups": rules_to_groups,
},
)
if random.random() < 0.01:
logger.info(
"delayed_processing.rule_to_fire",
extra={"rules_to_fire": list(rules_to_fire.keys()), "project_id": project_id},
)

parsed_rulegroup_to_event_data = parse_rulegroup_to_event_data(rulegroup_to_event_data)
with metrics.timer("delayed_processing.fire_rules.duration"):
fire_rules(rules_to_fire, parsed_rulegroup_to_event_data, alert_rules, project)

cleanup_redis_buffer(project, rules_to_groups, batch_key)
with sentry_sdk.start_span(
op="delayed_processing.fire_rules", name="Fire rules in delayed processing"
):
parsed_rulegroup_to_event_data = parse_rulegroup_to_event_data(rulegroup_to_event_data)
with metrics.timer("delayed_processing.fire_rules.duration"):
fire_rules(rules_to_fire, parsed_rulegroup_to_event_data, alert_rules, project)

with sentry_sdk.start_span(
op="delayed_processing.cleanup_redis_buffer",
name="Clean up redis buffer in delayed processing",
):
cleanup_redis_buffer(project, rules_to_groups, batch_key)


@delayed_processing_registry.register("delayed_processing") # default delayed processing
Expand Down
Loading