feat(aci): Allow equations in span metric monitors behind a flag

src/sentry/features/temporary.py

@@ -245,6 +245,8 @@ def register_temporary_features(manager: FeatureManager) -> None:
     manager.add("organizations:preprod-size-monitors-frontend", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable duplicating monitors from their details page
     manager.add("organizations:monitor-duplication", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
+    # Enable equation aggregates (equation|...) in span metric monitors
+    manager.add("organizations:spans-monitor-equations", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enables the Heroku log drain endpoint in relay
     manager.add("organizations:relay-heroku-log-drain-endpoint", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enables the playstation ingestion in relay

src/sentry/incidents/charts.py

@@ -210,8 +210,8 @@ def build_metric_alert_chart(
         ),
     }
 
-    # translate_aggregate_field can't resolve equations. Only trace metric queries
-    # accept them, and the validator never translates those aggregates.
+    # translate_aggregate_field can't resolve equations. Only trace metric and span
+    # queries accept them, and the validator never translates those aggregates.
     aggregate = (
         snuba_query.aggregate
         if is_equation(snuba_query.aggregate)

src/sentry/search/eap/spans/validator.py

@@ -0,0 +1,61 @@
+from datetime import timedelta
+
+from django.utils import timezone
+from django.utils.translation import gettext_lazy as _
+from rest_framework import serializers
+
+from sentry.discover.arithmetic import ArithmeticError, parse_arithmetic, strip_equation
+from sentry.exceptions import InvalidSearchQuery
+from sentry.incidents.logic import EAP_FUNCTIONS
+from sentry.models.organization import Organization
+from sentry.models.project import Project
+from sentry.search.eap.resolver import SearchResolver
+from sentry.search.eap.spans.definitions import SPAN_DEFINITIONS
+from sentry.search.eap.types import SearchResolverConfig
+from sentry.search.events.fields import is_function
+from sentry.search.events.types import SnubaParams
+
+
+def validate_spans_equation_aggregate(
+    aggregate: str, organization: Organization, projects: list[Project]
+) -> None:
+    equation = strip_equation(aggregate)
+    try:
+        _operation, fields, functions = parse_arithmetic(equation, definitions=SPAN_DEFINITIONS)
+    except ArithmeticError as e:
+        raise serializers.ValidationError({"aggregate": _(f"Invalid Metric: {e}")})
+
+    # The subscription only sends aggregate expressions to Snuba, so an equation without
+    # functions would produce an empty request.
+    if fields or not functions:
+        raise serializers.ValidationError(
+            {
+                "aggregate": _(
+                    "Invalid Metric: An equation must combine aggregate functions, such as "
+                    "sum(a) / sum(b)."
+                )
+            }
+        )
+
+    for function in functions:
+        match = is_function(function)
+        if match is None or match.group("function") not in EAP_FUNCTIONS:
+            raise serializers.ValidationError(
+                {"aggregate": _(f"Invalid Metric: {function} is not supported in monitors.")}
+            )
+
+    now = timezone.now()
+    resolver = SearchResolver(
+        # Formulas such as opportunity_score run a query while resolving, so scope it to the
+        # monitor's organization and projects.
+        params=SnubaParams(
+            projects=projects, organization=organization, start=now - timedelta(hours=1), end=now
+        ),
+        config=SearchResolverConfig(),
+        definitions=SPAN_DEFINITIONS,
+    )
+    try:
+        resolver.resolve_equation(equation)
+    # Number arguments are converted with float() and int(), so a bad value raises ValueError.
+    except (InvalidSearchQuery, ArithmeticError, ValueError) as e:
+        raise serializers.ValidationError({"aggregate": _(f"Invalid Metric: {e}")})

src/sentry/snuba/snuba_query_validator.py

@@ -10,6 +10,7 @@
 from snuba_sdk import Column, Condition, Entity, Limit, Op
 
 from sentry import features
+from sentry.discover.arithmetic import is_equation
 from sentry.exceptions import (
     IncompatibleMetricsQuery,
     InvalidSearchQuery,
@@ -26,6 +27,7 @@
 from sentry.models.environment import Environment
 from sentry.models.project import Project
 from sentry.search.eap.constants import VALID_GRANULARITIES
+from sentry.search.eap.spans.validator import validate_spans_equation_aggregate
 from sentry.search.eap.trace_metrics.validator import validate_trace_metrics_aggregate
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.entity_subscription import (
@@ -268,8 +270,26 @@ def _validate_aggregate(self, data: dict[str, Any]) -> None:
             and SnubaQueryEventType.EventType.TRACE_ITEM_METRIC in event_types
         )
 
+        is_spans = (
+            dataset == Dataset.EventsAnalyticsPlatform
+            and event_types
+            and SnubaQueryEventType.EventType.TRACE_ITEM_SPAN in event_types
+        )
+
         if is_trace_metrics:
             validate_trace_metrics_aggregate(aggregate)
+        elif is_spans and is_equation(aggregate):
+            if not features.has(
+                "organizations:spans-monitor-equations",
+                self.context["organization"],
+                actor=self.context.get("user", None),
+            ):
+                raise serializers.ValidationError(
+                    {"aggregate": _("Invalid Metric: Span monitors do not support equations.")}
+                )
+            validate_spans_equation_aggregate(
+                aggregate, self.context["organization"], [self.context["project"]]
+            )
         else:
             try:
                 if not check_aggregate_column_support(

tests/sentry/snuba/test_entity_subscriptions.py

@@ -1,4 +1,5 @@
 import pytest
+from sentry_protos.snuba.v1.endpoint_time_series_pb2 import Expression
 from sentry_protos.snuba.v1.trace_item_attribute_pb2 import (
     ExtrapolationMode as ProtoExtrapolationMode,
 )
@@ -433,3 +434,35 @@ def test_get_entity_subscription_for_eap_with_extrapolation_mode(self) -> None:
             rpc_timeseries_request.expressions[0].aggregation.extrapolation_mode
             == ProtoExtrapolationMode.EXTRAPOLATION_MODE_SERVER_ONLY
         )
+
+    def test_get_entity_subscription_for_eap_equation(self) -> None:
+        aggregate = "equation|sum(span.duration) / count(span.duration)"
+        entity_subscription = get_entity_subscription(
+            query_type=SnubaQuery.Type.PERFORMANCE,
+            dataset=Dataset.EventsAnalyticsPlatform,
+            aggregate=aggregate,
+            time_window=3600,
+            extra_fields={"org_id": self.organization.id},
+        )
+        assert isinstance(entity_subscription, PerformanceSpansEAPRpcEntitySubscription)
+
+        rpc_timeseries_request = entity_subscription.build_rpc_request(
+            "span.op:http.client", [self.project.id], None
+        )
+
+        # Snuba subscriptions require exactly one expression with an extrapolation mode on
+        # every aggregation, so the equation must arrive as a single formula.
+        assert len(rpc_timeseries_request.expressions) == 1
+        expression = rpc_timeseries_request.expressions[0]
+        assert expression.label == aggregate
+        assert expression.formula.op == Expression.BinaryFormula.OP_DIVIDE
+        assert expression.formula.left.aggregation.label == "sum(span.duration)"
+        assert expression.formula.right.aggregation.label == "count(span.duration)"
+        assert (
+            expression.formula.left.aggregation.extrapolation_mode
+            == ProtoExtrapolationMode.EXTRAPOLATION_MODE_SAMPLE_WEIGHTED
+        )
+        assert (
+            expression.formula.right.aggregation.extrapolation_mode
+            == ProtoExtrapolationMode.EXTRAPOLATION_MODE_SAMPLE_WEIGHTED
+        )

tests/sentry/snuba/test_validators.py

@@ -198,6 +198,85 @@ def test_eap_unparseable_equation_aggregate(self) -> None:
         assert not validator.is_valid()
         assert "aggregate" in validator.errors
 
+    def _spans_equation_data(self, aggregate: str) -> dict[str, object]:
+        return {
+            "dataset": Dataset.EventsAnalyticsPlatform.value,
+            "query": "span.op:gen_ai.chat",
+            "aggregate": aggregate,
+            "timeWindow": 3600,
+            "environment": self.environment.name,
+            "eventTypes": [SnubaQueryEventType.EventType.TRACE_ITEM_SPAN.name.lower()],
+        }
+
+    @with_feature("organizations:spans-monitor-equations")
+    def test_eap_spans_equation_aggregate(self) -> None:
+        for aggregate in [
+            "equation|sum(gen_ai.usage.cache_read.input_tokens) / sum(gen_ai.usage.input_tokens)",
+            "equation|opportunity_score(measurements.score.lcp) * 100",
+        ]:
+            with self.subTest(aggregate=aggregate):
+                validator = SnubaQueryValidator(
+                    data=self._spans_equation_data(aggregate), context=self.context
+                )
+                assert validator.is_valid(), validator.errors
+                assert validator.validated_data["aggregate"] == aggregate
+
+    def test_eap_spans_equation_aggregate_without_feature(self) -> None:
+        validator = SnubaQueryValidator(
+            data=self._spans_equation_data("equation|count() / count(span.duration)"),
+            context=self.context,
+        )
+        assert not validator.is_valid()
+        assert validator.errors["aggregate"] == [
+            ErrorDetail(
+                string="Invalid Metric: Span monitors do not support equations.", code="invalid"
+            )
+        ]
+
+    @with_feature("organizations:spans-monitor-equations")
+    def test_eap_spans_equation_invalid(self) -> None:
+        for aggregate, error in [
+            ("equation|count_sample() / count()", "count_sample() is not supported in monitors."),
+            (
+                "equation|span.duration / 1000",
+                "An equation must combine aggregate functions, such as sum(a) / sum(b).",
+            ),
+            (
+                "equation|sum(span.duration) / span.self_time",
+                "Cannot mix functions and fields in arithmetic",
+            ),
+            (
+                "equation|sum(span.duration) /",
+                "Unable to parse your equation, make sure it is well formed arithmetic",
+            ),
+            ("equation|sum(span.op) / count()", "span.op is invalid for parameter 1 in sum"),
+            ("equation|apdex(span.duration,foo) * 2", "could not convert string to float: 'foo'"),
+        ]:
+            with self.subTest(aggregate=aggregate):
+                validator = SnubaQueryValidator(
+                    data=self._spans_equation_data(aggregate), context=self.context
+                )
+                assert not validator.is_valid()
+                assert str(validator.errors["aggregate"][0]).startswith(f"Invalid Metric: {error}")
+
+    @with_feature(["organizations:ourlogs-enabled", "organizations:spans-monitor-equations"])
+    def test_eap_logs_equation_aggregate(self) -> None:
+        validator = SnubaQueryValidator(
+            data={
+                **self._spans_equation_data("equation|count() / count(message)"),
+                "query": "",
+                "eventTypes": [SnubaQueryEventType.EventType.TRACE_ITEM_LOG.name.lower()],
+            },
+            context=self.context,
+        )
+        assert not validator.is_valid()
+        assert validator.errors["aggregate"] == [
+            ErrorDetail(
+                string="Invalid Metric: equation|count() / count(message) is not a valid function",
+                code="invalid",
+            )
+        ]
+
     def test_eap_performance_score_aggregate(self) -> None:
         data = {
             "dataset": Dataset.EventsAnalyticsPlatform.value,

tests/sentry/workflow_engine/endpoints/test_organization_project_detector_index.py

@@ -655,6 +655,57 @@ def test_use_metrics_equation_as_aggregate(self) -> None:
             SnubaQueryEventType.EventType.TRACE_ITEM_METRIC
         ]
 
+    def _spans_equation_data(self, equation: str) -> dict[str, Any]:
+        return {
+            **self.valid_data,
+            "dataSources": [
+                {
+                    "queryType": SnubaQuery.Type.PERFORMANCE.value,
+                    "dataset": Dataset.EventsAnalyticsPlatform.value,
+                    "query": "span.op:gen_ai.chat",
+                    "aggregate": equation,
+                    "timeWindow": 3600,
+                    "environment": self.environment.name,
+                    "eventTypes": [SnubaQueryEventType.EventType.TRACE_ITEM_SPAN.name.lower()],
+                }
+            ],
+        }
+
+    @with_feature("organizations:spans-monitor-equations")
+    def test_use_spans_equation_as_aggregate(self) -> None:
+        equation = (
+            "equation|sum(gen_ai.usage.cache_read.input_tokens) / sum(gen_ai.usage.input_tokens)"
+        )
+
+        with self.tasks():
+            response = self.get_success_response(
+                self.organization.slug,
+                self.project.slug,
+                **self._spans_equation_data(equation),
+                status_code=201,
+            )
+
+        assert response.data["dataSources"][0]["queryObj"]["snubaQuery"]["aggregate"] == equation
+
+        detector = Detector.objects.get(id=response.data["id"])
+        data_source = DataSource.objects.get(detector=detector)
+        query_sub = QuerySubscription.objects.get(id=int(data_source.source_id))
+        assert query_sub.status == QuerySubscription.Status.ACTIVE.value
+        assert query_sub.subscription_id is not None
+        assert query_sub.snuba_query.dataset == Dataset.EventsAnalyticsPlatform.value
+        assert query_sub.snuba_query.query == "span.op:gen_ai.chat"
+        assert query_sub.snuba_query.aggregate == equation
+        assert query_sub.snuba_query.event_types == [SnubaQueryEventType.EventType.TRACE_ITEM_SPAN]
+
+    def test_use_spans_equation_as_aggregate_without_feature(self) -> None:
+        response = self.get_error_response(
+            self.organization.slug,
+            self.project.slug,
+            **self._spans_equation_data("equation|count() / count(span.duration)"),
+            status_code=400,
+        )
+        assert "Span monitors do not support equations" in str(response.data)
+
 
 @cell_silo_test
 class OrganizationProjectDetectorIndexMonitorPostTest(APITestCase):

tests/sentry/workflow_engine/endpoints/validators/test_metric_data_source_schema.py

@@ -130,6 +130,9 @@ def test_accepted_aggregates(data_source: dict[str, Any], title: str) -> None:
         ),
         _data_source("events_analytics_platform", ["trace_item_log"], "p100(severity)"),
         _data_source("events_analytics_platform", ["trace_item_log"], "failure_rate()"),
+        _data_source(
+            "events_analytics_platform", ["trace_item_log"], "equation|count() / count(message)"
+        ),
         _data_source("metrics", [], "count()"),
     ],
 )