feat(on_demand): Support epm and eps operators

src/sentry/snuba/metrics/extraction.py

@@ -121,6 +121,7 @@
     "failure_count": "on_demand_failure_count",
     "failure_rate": "on_demand_failure_rate",
     "apdex": "on_demand_apdex",
+    "epm": "on_demand_epm",
 }
 
 # Mapping to infer metric type from Discover function.
@@ -137,6 +138,7 @@
     "failure_count": "c",
     "failure_rate": "c",
     "apdex": "c",
+    "epm": "c",
 }
 
 # Query fields that on their own do not require on-demand metric extraction but if present in an on-demand query
@@ -620,11 +622,16 @@ def apdex_tag_spec(project: Project, argument: Optional[str]) -> List[TagSpec]:
     ]
 
 
+def epm_tag_spec(_1: Project, _2: Optional[str]) -> List[TagSpec]:
+    return []
+
+
 # This is used to map a metric to a function which generates a specification
 _DERIVED_METRICS: Dict[MetricOperationType, TagsSpecsGenerator] = {
     "on_demand_failure_count": failure_tag_spec,
     "on_demand_failure_rate": failure_tag_spec,
     "on_demand_apdex": apdex_tag_spec,
+    "on_demand_epm": epm_tag_spec,
 }
 
 
@@ -700,7 +707,7 @@ def _field_for_hash(self) -> Optional[str]:
         # 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_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,7 @@
     min_timestamp,
     miserable_users,
     on_demand_apdex_snql_factory,
+    on_demand_epm_snql_factory,
     on_demand_failure_count_snql_factory,
     on_demand_failure_rate_snql_factory,
     rate_snql_factory,
@@ -1785,21 +1786,27 @@ def generate_where_statements(
         ),
         # Custom operations used for on demand derived metrics.
         DerivedOp(
-            op="on_demand_failure_count",
+            op="on_demand_apdex",
             can_orderby=True,
-            snql_func=on_demand_failure_count_snql_factory,
+            snql_func=on_demand_apdex_snql_factory,
             default_null_value=0,
         ),
         DerivedOp(
-            op="on_demand_failure_rate",
+            op="on_demand_epm",
             can_orderby=True,
-            snql_func=on_demand_failure_rate_snql_factory,
+            snql_func=on_demand_epm_snql_factory,
             default_null_value=0,
         ),
         DerivedOp(
-            op="on_demand_apdex",
+            op="on_demand_failure_count",
             can_orderby=True,
-            snql_func=on_demand_apdex_snql_factory,
+            snql_func=on_demand_failure_count_snql_factory,
+            default_null_value=0,
+        ),
+        DerivedOp(
+            op="on_demand_failure_rate",
+            can_orderby=True,
+            snql_func=on_demand_failure_rate_snql_factory,
             default_null_value=0,
         ),
     ]

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

@@ -802,22 +802,28 @@ def max_timestamp(aggregate_filter, org_id, use_case_id, alias=None):
     return timestamp_column_snql("maxIf", aggregate_filter, org_id, use_case_id, alias)
 
 
-def on_demand_failure_rate_snql_factory(aggregate_filter, org_id, use_case_id, alias=None):
+def total_count(aggregate_filter: Function) -> Function:
+    return Function("sumIf", [Column("value"), aggregate_filter])
+
+
+def on_demand_failure_rate_snql_factory(
+    aggregate_filter: Function, org_id: int, use_case_id: UseCaseID, alias: Optional[str]
+):
     """Divide the number of transactions that failed from the total."""
     return Function(
         "divide",
         [
             on_demand_failure_count_snql_factory(
                 aggregate_filter, org_id, use_case_id, "failure_count"
             ),
-            Function("sumIf", [Column("value"), aggregate_filter]),
+            total_count(aggregate_filter),
         ],
         alias=alias,
     )
 
 
 def on_demand_failure_count_snql_factory(
-    aggregate_filter: Function, org_id: int, use_case_id: UseCaseID, alias: str
+    aggregate_filter: Function, org_id: int, use_case_id: UseCaseID, alias: Optional[str]
 ) -> Function:
     """Count the number of transactions where the failure tag is set to true."""
     return Function(
@@ -842,7 +848,9 @@ def on_demand_failure_count_snql_factory(
     )
 
 
-def on_demand_apdex_snql_factory(aggregate_filter, org_id, use_case_id, alias=None):
+def on_demand_apdex_snql_factory(
+    aggregate_filter: Function, org_id: int, use_case_id: UseCaseID, alias: Optional[str]
+):
     # For more information about the formula, check https://docs.sentry.io/product/performance/metrics/#apdex.
 
     satisfactory = Function(
@@ -889,10 +897,17 @@ def on_demand_apdex_snql_factory(aggregate_filter, org_id, use_case_id, alias=No
             2,
         ],
     )
-    total = Function("sumIf", [Column("value"), aggregate_filter])
 
     return Function(
         "divide",
-        [Function("plus", [satisfactory, tolerable_divided_by_2]), total],
+        [Function("plus", [satisfactory, tolerable_divided_by_2]), total_count(aggregate_filter)],
         alias=alias,
     )
+
+
+def on_demand_epm_snql_factory(
+    aggregate_filter: Function, org_id: int, use_case_id: UseCaseID, alias: Optional[str]
+) -> Function:
+    """Return the count based on the aggregation."""
+    # Dividing by the time interval happens in a different place
+    return total_count(aggregate_filter)

src/sentry/snuba/metrics/utils.py

@@ -286,9 +286,10 @@ class MetricMetaWithTagKeys(MetricMeta):
     "min_timestamp",
     "max_timestamp",
     # Custom operations used for on demand derived metrics.
+    "on_demand_apdex",
+    "on_demand_epm",
     "on_demand_failure_count",
     "on_demand_failure_rate",
-    "on_demand_apdex",
 )
 OPERATIONS = (
     (

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
+def test_get_metric_extraction_config_with_epm(default_project):
+    with Feature({ON_DEMAND_METRICS: True, ON_DEMAND_METRICS_WIDGETS: True}):
+        create_widget(["epm()"], "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",
+            # XXX: What should the tags be??
+            "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,45 @@ def test_run_query_with_on_demand_apdex(self):
             ],
         )
 
+    # XXX: This is not correct since we're not yet dividing by the interval
+    def test_run_query_with_on_demand_epm(self):
+        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,
+        )
+        # XXX: Store a second later and then 61 seconds later
+        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")
+        # XXX: How is this 0 rather than 2?
+        assert result["data"][:1] == [{"time": timestamp.isoformat(), "epm": 0}]
+        assert result["meta"] == [
+            {"name": "time", "type": "DateTime('Universal')"},
+            # {"name": "epm", "type": "Float64"}, ???
+        ]
+
 
 class HistogramMetricQueryBuilderTest(MetricBuilderBaseTest):
     def test_histogram_columns_set_on_builder(self):

tests/sentry/snuba/test_extraction.py

@@ -8,6 +8,7 @@
     OnDemandMetricSpec,
     apdex_tag_spec,
     cleanup_query,
+    epm_tag_spec,
     failure_tag_spec,
     query_tokens_to_string,
     should_use_on_demand_metrics,
@@ -282,6 +283,17 @@ 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) == epm_tag_spec(default_project, "not_used")
+
+
 def test_cleanup_equivalent_specs():
     simple_spec = OnDemandMetricSpec("count()", "transaction.duration:>0")
     event_type_spec = OnDemandMetricSpec(