feat(rca): Add epm_by_timestamp to span metrics (#68178)

Add epm_by_timestamp to span metrics to calculate throughput of a span
before and after a breakpoint.

src/sentry/search/events/datasets/spans_metrics.py

@@ -507,6 +507,17 @@ class SpansMetricsDatasetConfig(DatasetConfig):
                     ),
                     default_result_type="duration",
                 ),
+                fields.MetricsFunction(
+                    "epm_by_timestamp",
+                    required_args=[
+                        fields.SnQLStringArg("condition", allowed_strings=["greater", "less"]),
+                        fields.TimestampArg("timestamp"),
+                    ],
+                    snql_distribution=lambda args, alias: self._resolve_epm_condition(
+                        args, args["condition"], alias
+                    ),
+                    default_result_type="rate",
+                ),
             ]
         }
 
@@ -866,6 +877,7 @@ class SpansMetricsDatasetConfig(DatasetConfig):
         self,
         args: Mapping[str, str | Column | SelectType | int | float | datetime],
         condition: str,
+        alias: str | None = None,
     ) -> SelectType:
         if condition == "greater":
             interval = (self.builder.params.end - args["timestamp"]).total_seconds()
@@ -891,6 +903,7 @@ class SpansMetricsDatasetConfig(DatasetConfig):
                 ),
                 Function("divide", [interval, 60]),
             ],
+            alias,
         )
 
     def _resolve_avg_condition(

tests/snuba/api/endpoints/test_organization_events_span_metrics.py

@@ -1279,6 +1279,61 @@ class OrganizationEventsMetricsEnhancedPerformanceEndpointTest(MetricsEnhancedPe
             == "avg_by_timestamp: condition argument invalid: string must be one of ['greater', 'less']"
         )
 
+    def test_epm_by_timestamp(self):
+        self.store_span_metric(
+            1,
+            internal_metric=constants.SELF_TIME_LIGHT,
+            timestamp=self.six_min_ago,
+            tags={},
+        )
+
+        # More events occur after the timestamp
+        for _ in range(3):
+            self.store_span_metric(
+                3,
+                internal_metric=constants.SELF_TIME_LIGHT,
+                timestamp=self.min_ago,
+                tags={},
+            )
+
+        response = self.do_request(
+            {
+                "field": [
+                    f"epm_by_timestamp(less,{int(self.two_min_ago.timestamp())})",
+                    f"epm_by_timestamp(greater,{int(self.two_min_ago.timestamp())})",
+                ],
+                "query": "",
+                "project": self.project.id,
+                "dataset": "spansMetrics",
+                "statsPeriod": "1h",
+            }
+        )
+
+        assert response.status_code == 200, response.content
+        data = response.data["data"]
+        assert len(data) == 1
+        assert data[0][f"epm_by_timestamp(less,{int(self.two_min_ago.timestamp())})"] < 1.0
+        assert data[0][f"epm_by_timestamp(greater,{int(self.two_min_ago.timestamp())})"] > 1.0
+
+    def test_epm_by_timestamp_invalid_condition(self):
+        response = self.do_request(
+            {
+                "field": [
+                    f"epm_by_timestamp(INVALID_ARG,{int(self.two_min_ago.timestamp())})",
+                ],
+                "query": "",
+                "project": self.project.id,
+                "dataset": "spansMetrics",
+                "statsPeriod": "1h",
+            }
+        )
+
+        assert response.status_code == 400, response.content
+        assert (
+            response.data["detail"]
+            == "epm_by_timestamp: condition argument invalid: string must be one of ['greater', 'less']"
+        )
+
 
 class OrganizationEventsMetricsEnhancedPerformanceEndpointTestWithMetricLayer(
     OrganizationEventsMetricsEnhancedPerformanceEndpointTest