Merge branch 'master' into ryan953/ref-replay-hydration-diff-loading

Author: ryan953

src/sentry/api/endpoints/organization_release_meta.py

@@ -104,7 +104,7 @@ def get(self, request: Request, organization, version) -> Response:
                 "version": release.version,
                 "versionInfo": expose_version_info(release.version_info),
                 "projects": projects,
-                "newGroups": release.new_groups,
+                "newGroups": sum(project["newGroups"] for project in projects),
                 "deployCount": release.total_deploys,
                 "commitCount": release.commit_count,
                 "released": release.date_released or release.date_added,

src/sentry/api/endpoints/project_seer_preferences.py

@@ -54,7 +54,9 @@ class ProjectSeerPreferencesEndpoint(ProjectEndpoint):
     }
 
     def post(self, request: Request, project: Project) -> Response:
-        if not features.has("organizations:autofix-seer-preferences", project.organization):
+        if not features.has(
+            "organizations:autofix-seer-preferences", project.organization, actor=request.user
+        ):
             return Response("Feature flag not enabled", status=403)
 
         data = orjson.loads(request.body)
@@ -86,7 +88,9 @@ def post(self, request: Request, project: Project) -> Response:
         return Response(status=204)
 
     def get(self, request: Request, project: Project) -> Response:
-        if not features.has("organizations:autofix-seer-preferences", project.organization):
+        if not features.has(
+            "organizations:autofix-seer-preferences", project.organization, actor=request.user
+        ):
             return Response("Feature flag not enabled", status=403)
 
         path = "/v1/project-preference"

src/sentry/conf/server.py

@@ -3025,7 +3025,6 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
     "buffered-segments": "default",
     "buffered-segments-dlq": "default",
     "taskworker": "default",
-    "task-worker": "default",
     "snuba-ourlogs": "default",
 }
 

src/sentry/conf/types/kafka_definition.py

@@ -65,7 +65,6 @@ class Topic(Enum):
     BUFFERED_SEGMENTS = "buffered-segments"
     BUFFERED_SEGMENTS_DLQ = "buffered-segments-dlq"
     TASKWORKER = "taskworker"
-    TASK_WORKER = "task-worker"
 
 
 class ConsumerDefinition(TypedDict, total=False):

src/sentry/integrations/discord/actions/metric_alert.py

@@ -34,12 +34,16 @@ def send_incident_alert_notification(
     notification_context: NotificationContext,
     metric_issue_context: MetricIssueContext,
     open_period_context: OpenPeriodContext,
-    alert_rule_serialized_response: AlertRuleSerializerResponse,
-    incident_serialized_response: DetailedIncidentSerializerResponse,
+    alert_rule_serialized_response: AlertRuleSerializerResponse | None,
+    incident_serialized_response: DetailedIncidentSerializerResponse | None,
     notification_uuid: str | None = None,
 ) -> bool:
     chart_url = None
-    if features.has("organizations:metric-alert-chartcuterie", organization):
+    if (
+        features.has("organizations:metric-alert-chartcuterie", organization)
+        and alert_rule_serialized_response
+        and incident_serialized_response
+    ):
         try:
             chart_url = build_metric_alert_chart(
                 organization=organization,

src/sentry/integrations/msteams/utils.py

@@ -110,8 +110,8 @@ def send_incident_alert_notification(
     notification_context: NotificationContext,
     metric_issue_context: MetricIssueContext,
     open_period_context: OpenPeriodContext,
-    alert_rule_serialized_response: AlertRuleSerializerResponse,
-    incident_serialized_response: DetailedIncidentSerializerResponse,
+    alert_rule_serialized_response: AlertRuleSerializerResponse | None,
+    incident_serialized_response: DetailedIncidentSerializerResponse | None,
     notification_uuid: str | None = None,
 ) -> bool:
     from .card_builder.incident_attachment import build_incident_attachment

src/sentry/integrations/slack/utils/notifications.py

@@ -187,12 +187,16 @@ def _build_notification_payload(
     alert_context: AlertContext,
     metric_issue_context: MetricIssueContext,
     open_period_context: OpenPeriodContext,
-    alert_rule_serialized_response: AlertRuleSerializerResponse,
-    incident_serialized_response: DetailedIncidentSerializerResponse,
+    alert_rule_serialized_response: AlertRuleSerializerResponse | None,
+    incident_serialized_response: DetailedIncidentSerializerResponse | None,
     notification_uuid: str | None,
 ) -> tuple[str, str]:
     chart_url = None
-    if features.has("organizations:metric-alert-chartcuterie", organization):
+    if (
+        features.has("organizations:metric-alert-chartcuterie", organization)
+        and alert_rule_serialized_response
+        and incident_serialized_response
+    ):
         try:
             chart_url = build_metric_alert_chart(
                 organization=organization,
@@ -395,8 +399,8 @@ def send_incident_alert_notification(
     notification_context: NotificationContext,
     metric_issue_context: MetricIssueContext,
     open_period_context: OpenPeriodContext,
-    alert_rule_serialized_response: AlertRuleSerializerResponse,
-    incident_serialized_response: DetailedIncidentSerializerResponse,
+    alert_rule_serialized_response: AlertRuleSerializerResponse | None,
+    incident_serialized_response: DetailedIncidentSerializerResponse | None,
     notification_uuid: str | None = None,
 ) -> bool:
     # Make sure organization integration is still active:

src/sentry/tasks/base.py

@@ -10,6 +10,7 @@
 from celery import Task, current_task
 from django.conf import settings
 from django.db.models import Model
+from kombu import Producer
 
 from sentry import options
 from sentry.celery import app
@@ -80,6 +81,9 @@ def override(*args: P.args, **kwargs: P.kwargs) -> R:
 
         random.seed(datetime.now().timestamp())
         if rollout > random.random():
+            producer = kwargs.get("producer")
+            if isinstance(producer, Producer):
+                del kwargs["producer"]
             return taskworker_attr(*args, **kwargs)
 
         return celery_task_attr(*args, **kwargs)

src/sentry/taskworker/registry.py

@@ -132,9 +132,8 @@ def wrapped(func: Callable[P, R]) -> Task[P, R]:
     def _handle_produce_future(self, future: ProducerFuture, tags: dict[str, str]) -> None:
         if future.cancelled():
             metrics.incr("taskworker.registry.send_task.cancelled", tags=tags)
-        elif future.exception(
-            1
-        ):  # this does not block since this callback only gets run when the future is finished and exception is set
+        elif future.exception(1):
+            # this does not block since this callback only gets run when the future is finished and exception is set
             metrics.incr("taskworker.registry.send_task.failed", tags=tags)
         else:
             metrics.incr("taskworker.registry.send_task.success", tags=tags)

src/sentry/taskworker/worker.py

@@ -82,7 +82,7 @@ def child_worker(
     for module in settings.TASKWORKER_IMPORTS:
         __import__(module)
 
-    current_task_id: str | None = None
+    current_activation: TaskActivation | None = None
     processed_task_count = 0
 
     def handle_alarm(signum: int, frame: Any) -> None:
@@ -92,13 +92,21 @@ def handle_alarm(signum: int, frame: Any) -> None:
         If we hit an alarm in a child, we need to push a result
         and terminate the child.
         """
-        nonlocal current_task_id, processed_tasks
+        nonlocal current_activation, processed_tasks
 
-        if current_task_id:
+        if current_activation:
             processed_tasks.put(
-                ProcessingResult(task_id=current_task_id, status=TASK_ACTIVATION_STATUS_FAILURE)
+                ProcessingResult(
+                    task_id=current_activation.id, status=TASK_ACTIVATION_STATUS_FAILURE
+                )
+            )
+            metrics.incr(
+                "taskworker.worker.processing_deadline_exceeded",
+                tags={
+                    "namespace": current_activation.namespace,
+                    "taskname": current_activation.taskname,
+                },
             )
-        metrics.incr("taskworker.worker.processing_deadline_exceeded")
         sys.exit(1)
 
     signal.signal(signal.SIGALRM, handle_alarm)
@@ -137,15 +145,15 @@ def handle_alarm(signum: int, frame: Any) -> None:
             key = get_at_most_once_key(activation.namespace, activation.taskname, activation.id)
             if cache.add(key, "1", timeout=AT_MOST_ONCE_TIMEOUT):  # The key didn't exist
                 metrics.incr(
-                    "taskworker.task.at_most_once.executed", tags={"task": activation.taskname}
+                    "taskworker.task.at_most_once.executed", tags={"taskname": activation.taskname}
                 )
             else:
                 metrics.incr(
-                    "taskworker.worker.at_most_once.skipped", tags={"task": activation.taskname}
+                    "taskworker.worker.at_most_once.skipped", tags={"taskname": activation.taskname}
                 )
                 continue
 
-        current_task_id = activation.id
+        current_activation = activation
 
         # Set an alarm for the processing_deadline_duration
         signal.alarm(activation.processing_deadline_duration)
@@ -159,7 +167,7 @@ def handle_alarm(signum: int, frame: Any) -> None:
             signal.alarm(0)
         except Exception as err:
             if task_func.should_retry(activation.retry_state, err):
-                logger.info("taskworker.task.retry", extra={"task": activation.taskname})
+                logger.info("taskworker.task.retry", extra={"taskname": activation.taskname})
                 next_state = TASK_ACTIVATION_STATUS_RETRY
 
             if next_state != TASK_ACTIVATION_STATUS_RETRY:
@@ -199,12 +207,12 @@ def handle_alarm(signum: int, frame: Any) -> None:
         metrics.distribution(
             "taskworker.worker.execution_duration",
             execution_duration,
-            tags={"namespace": activation.namespace},
+            tags={"namespace": activation.namespace, "taskname": activation.taskname},
         )
         metrics.distribution(
             "taskworker.worker.execution_latency",
             execution_latency,
-            tags={"namespace": activation.namespace},
+            tags={"namespace": activation.namespace, "taskname": activation.taskname},
         )