feat(on_demand): Support epm and eps operators (#56598)

Fixes #56572

src/sentry/snuba/metrics/extraction.py

@@ -123,6 +123,8 @@ _SEARCH_TO_DERIVED_METRIC_AGGREGATES: Dict[str, MetricOperationType] = {
     "failure_count": "on_demand_failure_count",
     "failure_rate": "on_demand_failure_rate",
     "apdex": "on_demand_apdex",
+    "epm": "on_demand_epm",
+    "eps": "on_demand_eps",
 }
 
 # Mapping to infer metric type from Discover function.
@@ -139,6 +141,8 @@ _AGGREGATE_TO_METRIC_TYPE = {
     "failure_count": "c",
     "failure_rate": "c",
     "apdex": "c",
+    "epm": "c",
+    "eps": "c",
 }
 
 # Query fields that on their own do not require on-demand metric extraction but if present in an on-demand query
@@ -623,10 +627,12 @@ def apdex_tag_spec(project: Project, argument: Optional[str]) -> List[TagSpec]:
 
 
 # This is used to map a metric to a function which generates a specification
-_DERIVED_METRICS: Dict[MetricOperationType, TagsSpecsGenerator] = {
+_DERIVED_METRICS: Dict[MetricOperationType, TagsSpecsGenerator | None] = {
     "on_demand_failure_count": failure_tag_spec,
     "on_demand_failure_rate": failure_tag_spec,
     "on_demand_apdex": apdex_tag_spec,
+    "on_demand_epm": None,
+    "on_demand_eps": None,
 }
 
 
@@ -702,7 +708,12 @@ class OnDemandMetricSpec:
         # with condition `f` and this will create a problem, since we might already have data for the `count()` and when
         # `apdex()` is created in the UI, we will use that metric but that metric didn't extract in the past the tags
         # that are used for apdex calculation, effectively causing problems with the data.
-        if self.op in ["on_demand_failure_count", "on_demand_failure_rate"]:
+        if self.op in [
+            "on_demand_epm",
+            "on_demand_eps",
+            "on_demand_failure_count",
+            "on_demand_failure_rate",
+        ]:
             return self.op
         elif self.op == "on_demand_apdex":
             return f"{self.op}:{self._argument}"

src/sentry/snuba/metrics/fields/base.py

@@ -61,6 +61,8 @@ from sentry.snuba.metrics.fields.snql import (
     min_timestamp,
     miserable_users,
     on_demand_apdex_snql_factory,
+    on_demand_epm_snql_factory,
+    on_demand_eps_snql_factory,
     on_demand_failure_count_snql_factory,
     on_demand_failure_rate_snql_factory,
     rate_snql_factory,
@@ -1792,6 +1794,18 @@ DERIVED_OPS: Mapping[MetricOperationType, DerivedOp] = {
             snql_func=on_demand_apdex_snql_factory,
             default_null_value=0,
         ),
+        DerivedOp(
+            op="on_demand_epm",
+            can_orderby=True,
+            snql_func=on_demand_epm_snql_factory,
+            default_null_value=0,
+        ),
+        DerivedOp(
+            op="on_demand_eps",
+            can_orderby=True,
+            snql_func=on_demand_eps_snql_factory,
+            default_null_value=0,
+        ),
         DerivedOp(
             op="on_demand_failure_count",
             can_orderby=True,

src/sentry/snuba/metrics/fields/snql.py

@@ -609,7 +609,12 @@ def histogram_snql_factory(
     )
 
 
-def rate_snql_factory(aggregate_filter, numerator, denominator=1.0, alias=None):
+def rate_snql_factory(
+    aggregate_filter: Function,
+    numerator: float,
+    denominator: float = 1.0,
+    alias: Optional[str] = None,
+) -> Function:
     return Function(
         "divide",
         [
@@ -903,3 +908,19 @@ def on_demand_apdex_snql_factory(
         [Function("plus", [satisfactory, tolerable_divided_by_2]), total_count(aggregate_filter)],
         alias=alias,
     )
+
+
+def on_demand_epm_snql_factory(
+    aggregate_filter: Function,
+    interval: float,
+    alias: Optional[str],
+) -> Function:
+    return rate_snql_factory(aggregate_filter, interval, 60, alias)
+
+
+def on_demand_eps_snql_factory(
+    aggregate_filter: Function,
+    interval: float,
+    alias: Optional[str],
+) -> Function:
+    return rate_snql_factory(aggregate_filter, interval, 1, alias)

src/sentry/snuba/metrics/query_builder.py

@@ -1138,6 +1138,10 @@ class SnubaQueryBuilder:
                     params = self._alias_to_metric_field[field[2]].params
                 except KeyError:
                     params = None
+
+                # In order to support on demand metrics which require an interval (e.g. epm),
+                # we want to pass the interval down via params so we can pass it to the associated snql_factory
+                params = {"interval": self._metrics_query.interval, **(params or {})}
                 select += metric_field_obj.generate_select_statements(
                     projects=self._projects,
                     use_case_id=self._use_case_id,

src/sentry/snuba/metrics/utils.py

@@ -96,9 +96,11 @@ MetricOperationType = Literal[
     "min_timestamp",
     "max_timestamp",
     # Custom operations used for on demand derived metrics.
+    "on_demand_apdex",
+    "on_demand_epm",
+    "on_demand_eps",
     "on_demand_failure_count",
     "on_demand_failure_rate",
-    "on_demand_apdex",
 ]
 MetricUnit = Literal[
     "nanosecond",
@@ -287,6 +289,8 @@ DERIVED_OPERATIONS = (
     "max_timestamp",
     # Custom operations used for on demand derived metrics.
     "on_demand_apdex",
+    "on_demand_epm",
+    "on_demand_eps",
     "on_demand_failure_count",
     "on_demand_failure_rate",
 )

tests/sentry/relay/config/test_metric_extraction.py

@@ -432,6 +432,25 @@ def test_get_metric_extraction_config_with_apdex(default_project):
         }
 
 
+@django_db_all
+@pytest.mark.parametrize("metric", [("epm()"), ("eps()")])
+def test_get_metric_extraction_config_with_no_tag_spec(default_project, metric):
+    with Feature({ON_DEMAND_METRICS: True, ON_DEMAND_METRICS_WIDGETS: True}):
+        create_widget([metric], "transaction.duration:>=1000", default_project)
+
+        config = get_metric_extraction_config(default_project)
+
+        assert config
+        assert len(config["metrics"]) == 1
+        assert config["metrics"][0] == {
+            "category": "transaction",
+            "condition": {"name": "event.duration", "op": "gte", "value": 1000.0},
+            "field": None,
+            "mri": "c:transactions/on_demand@none",
+            "tags": [{"key": "query_hash", "value": ANY}],
+        }
+
+
 @django_db_all
 @pytest.mark.parametrize(
     "enabled_features, number_of_metrics",

tests/sentry/search/events/builder/test_metrics.py

@@ -2255,6 +2255,64 @@ class TimeseriesMetricQueryBuilderTest(MetricBuilderBaseTest):
             ],
         )
 
+    def test_run_query_with_on_demand_epm(self):
+        """Test events per minute for 1 event within an hour."""
+        field = "epm()"
+        query = "transaction.duration:>=100"
+        spec = OnDemandMetricSpec(field=field, query=query)
+        timestamp = self.start
+        self.store_transaction_metric(
+            value=1,
+            metric=TransactionMetricKey.COUNT_ON_DEMAND.value,
+            internal_metric=TransactionMRI.COUNT_ON_DEMAND.value,
+            entity="metrics_counters",
+            tags={"query_hash": spec.query_hash},
+            timestamp=timestamp,
+        )
+        query = TimeseriesMetricQueryBuilder(
+            self.params,
+            dataset=Dataset.PerformanceMetrics,
+            interval=3600,
+            query=query,
+            selected_columns=[field],
+            config=QueryBuilderConfig(on_demand_metrics_enabled=True),
+        )
+        result = query.run_query("test_query")
+        assert result["data"][:1] == [{"time": timestamp.isoformat(), "epm": 1 / 60}]
+        assert result["meta"] == [
+            {"name": "time", "type": "DateTime('Universal')"},
+            {"name": "epm", "type": "Float64"},
+        ]
+
+    def test_run_query_with_on_demand_eps(self):
+        """Test event per second for 1 event within an hour."""
+        field = "eps()"
+        query = "transaction.duration:>=100"
+        spec = OnDemandMetricSpec(field=field, query=query)
+        timestamp = self.start
+        self.store_transaction_metric(
+            value=1,
+            metric=TransactionMetricKey.COUNT_ON_DEMAND.value,
+            internal_metric=TransactionMRI.COUNT_ON_DEMAND.value,
+            entity="metrics_counters",
+            tags={"query_hash": spec.query_hash},
+            timestamp=timestamp,
+        )
+        query = TimeseriesMetricQueryBuilder(
+            self.params,
+            dataset=Dataset.PerformanceMetrics,
+            interval=3600,
+            query=query,
+            selected_columns=[field],
+            config=QueryBuilderConfig(on_demand_metrics_enabled=True),
+        )
+        result = query.run_query("test_query")
+        assert result["data"][:1] == [{"time": timestamp.isoformat(), "eps": 1 / 60 / 60}]
+        assert result["meta"] == [
+            {"name": "time", "type": "DateTime('Universal')"},
+            {"name": "eps", "type": "Float64"},
+        ]
+
 
 class HistogramMetricQueryBuilderTest(MetricBuilderBaseTest):
     def test_histogram_columns_set_on_builder(self):

tests/sentry/snuba/test_extraction.py

@@ -332,6 +332,28 @@ def test_spec_apdex(_get_apdex_project_transaction_threshold, default_project):
     assert spec.tags_conditions(default_project) == apdex_tag_spec(default_project, "10")
 
 
+@django_db_all
+def test_spec_epm(default_project):
+    spec = OnDemandMetricSpec("epm()", "transaction.duration:>1s")
+
+    assert spec._metric_type == "c"
+    assert spec.field_to_extract is None
+    assert spec.op == "on_demand_epm"
+    assert spec.condition == {"name": "event.duration", "op": "gt", "value": 1000.0}
+    assert spec.tags_conditions(default_project) == []
+
+
+@django_db_all
+def test_spec_eps(default_project):
+    spec = OnDemandMetricSpec("eps()", "transaction.duration:>1s")
+
+    assert spec._metric_type == "c"
+    assert spec.field_to_extract is None
+    assert spec.op == "on_demand_eps"
+    assert spec.condition == {"name": "event.duration", "op": "gt", "value": 1000.0}
+    assert spec.tags_conditions(default_project) == []
+
+
 def test_cleanup_equivalent_specs():
     simple_spec = OnDemandMetricSpec("count()", "transaction.duration:>0")
     event_type_spec = OnDemandMetricSpec(