feat(alerts): on demand metric timeseries support

src/sentry/api/endpoints/organization_events_stats.py

@@ -97,6 +97,7 @@ def get_features(self, organization: Organization, request: Request) -> Mapping[
             "organizations:mep-rollout-flag",
             "organizations:use-metrics-layer",
             "organizations:starfish-view",
+            "organizations:on-demand-metrics-extraction",
         ]
         batch_features = features.batch_has(
             feature_names,
@@ -182,6 +183,8 @@ def get(self, request: Request, organization: Organization) -> Response:
             allow_metric_aggregates = request.GET.get("preventMetricAggregates") != "1"
             sentry_sdk.set_tag("performance.metrics_enhanced", metrics_enhanced)
 
+        use_on_demand_metrics = request.GET.get("useOnDemandMetrics")
+
         def get_event_stats(
             query_columns: Sequence[str],
             query: str,
@@ -217,6 +220,8 @@ def get_event_stats(
                 allow_metric_aggregates=allow_metric_aggregates,
                 has_metrics=use_metrics,
                 use_metrics_layer=batch_features.get("organizations:use-metrics-layer", False),
+                on_demand_metrics_enabled=use_on_demand_metrics
+                and batch_features.get("organizations:on-demand-metrics-extraction", False),
             )
 
         try:

src/sentry/search/events/builder/discover.py

@@ -219,6 +219,7 @@ def __init__(
         # Currently this is only used for avoiding conflicting values when doing the first query
         # of a top events request
         skip_tag_resolution: bool = False,
+        on_demand_metrics_enabled: bool = False,
     ):
         self.dataset = dataset
 
@@ -234,6 +235,7 @@ def __init__(
         self.transform_alias_to_input_format = transform_alias_to_input_format
         self.raw_equations = equations
         self.use_metrics_layer = use_metrics_layer
+        self.on_demand_metrics_enabled = on_demand_metrics_enabled
         self.auto_fields = auto_fields
         self.query = query
         self.groupby_columns = groupby_columns

src/sentry/search/events/builder/metrics.py

@@ -50,6 +50,7 @@
 class MetricsQueryBuilder(QueryBuilder):
     requires_organization_condition = True
     is_alerts_query = False
+    is_timeseries_query = False
     organization_column: str = "organization_id"
 
     def __init__(
@@ -75,10 +76,6 @@ def __init__(
         kwargs["has_metrics"] = True
         assert dataset is None or dataset in [Dataset.PerformanceMetrics, Dataset.Metrics]
 
-        self._on_demand_spec = self._resolve_on_demand_spec(
-            dataset, kwargs.get("selected_columns", []), kwargs.get("query", "")
-        )
-
         if granularity is not None:
             self._granularity = granularity
         super().__init__(
@@ -88,6 +85,10 @@ def __init__(
             *args,
             **kwargs,
         )
+        self._on_demand_spec = self._resolve_on_demand_spec(
+            dataset, kwargs.get("selected_columns", []), kwargs.get("query", "")
+        )
+
         org_id = self.filter_params.get("organization_id")
         if org_id is None and self.params.organization is not None:
             org_id = self.params.organization.id
@@ -98,8 +99,11 @@ def __init__(
     def _resolve_on_demand_spec(
         self, dataset: Optional[Dataset], selected_cols: List[Optional[str]], query: str
     ) -> Optional[OndemandMetricSpec]:
+        if not self.on_demand_metrics_enabled:
+            return None
+
         field = selected_cols[0] if selected_cols else None
-        if not self.is_alerts_query or not field:
+        if not field:
             return None
 
         if not is_on_demand_query(dataset, field, query):
@@ -112,30 +116,32 @@ def _resolve_on_demand_spec(
             return None
 
     def _get_on_demand_metrics_query(self) -> Optional[MetricsQuery]:
-        if not self.is_performance or not self.is_alerts_query:
-            return None
-
         spec = self._on_demand_spec
 
+        # TimeseriesQueryBuilder specific parameters
+        if self.is_timeseries_query:
+            limit = None
+            alias = "count"
+        else:
+            limit = self.limit
+            alias = spec.mri
+
         return MetricsQuery(
-            select=[MetricField(spec.op, spec.mri, alias=spec.mri)],
+            select=[MetricField(spec.op, spec.mri, alias=alias)],
             where=[
                 Condition(
                     lhs=Column(QUERY_HASH_KEY),
                     op=Op.EQ,
                     rhs=spec.query_hash(),
                 ),
             ],
-            # TODO(ogi): groupby and orderby
-            limit=self.limit,
+            limit=limit,
             offset=self.offset,
             granularity=self.resolve_granularity(),
             is_alerts_query=self.is_alerts_query,
             org_id=self.params.organization.id,
             project_ids=[p.id for p in self.params.projects],
-            # We do not need the series here, as later, we only extract the totals and assign it to the
-            # request.query
-            include_series=False,
+            include_series=True,
             start=self.params.start,
             end=self.params.end,
         )
@@ -198,7 +204,7 @@ def resolve_column_name(self, col: str) -> str:
             col = tag_match.group("tag") if tag_match else col
 
         # on-demand metrics require metrics layer behavior
-        if self.use_metrics_layer or self._on_demand_spec:
+        if self.use_metrics_layer or self.on_demand_metrics_enabled:
             if col in ["project_id", "timestamp"]:
                 return col
             # TODO: update resolve params so this isn't needed
@@ -571,7 +577,7 @@ def get_metrics_layer_snql_query(self) -> Request:
         snuba SDK.
         """
 
-        if not self.use_metrics_layer and not self._on_demand_spec:
+        if not self.use_metrics_layer and not self.on_demand_metrics_enabled:
             # The reasoning for this error is because if "use_metrics_layer" is false, the MQB will not generate the
             # snql dialect explained below as there is not need for that because it will directly generate normal snql
             # that can be returned via the "get_snql_query" method.
@@ -784,25 +790,24 @@ def validate_orderby_clause(self) -> None:
                 raise IncompatibleMetricsQuery("Can't orderby tags")
 
     def run_query(self, referrer: str, use_cache: bool = False) -> Any:
-        if self.use_metrics_layer or self._on_demand_spec:
+        if self.use_metrics_layer or self.on_demand_metrics_enabled:
             from sentry.snuba.metrics.datasource import get_series
             from sentry.snuba.metrics.mqb_query_transformer import (
                 transform_mqb_query_to_metrics_query,
             )
 
             try:
                 with sentry_sdk.start_span(op="metric_layer", description="transform_query"):
-                    if self._on_demand_spec:
-                        metric_query = self._get_on_demand_metrics_query()
+                    if self.on_demand_metrics_enabled:
+                        metrics_query = self._get_on_demand_metrics_query()
                     else:
-                        metric_query = transform_mqb_query_to_metrics_query(
+                        metrics_query = transform_mqb_query_to_metrics_query(
                             self.get_metrics_layer_snql_query().query, self.is_alerts_query
                         )
-                    # metric_query.where = metric_query_ondemand.where
                 with sentry_sdk.start_span(op="metric_layer", description="run_query"):
                     metrics_data = get_series(
                         projects=self.params.projects,
-                        metrics_query=metric_query,
+                        metrics_query=metrics_query,
                         use_case_id=UseCaseID.TRANSACTIONS
                         if self.is_performance
                         else UseCaseID.SESSIONS,
@@ -822,7 +827,7 @@ def run_query(self, referrer: str, use_cache: bool = False) -> Any:
                     data.update(group["totals"])
                     metric_layer_result["data"].append(data)
                     for meta in metric_layer_result["meta"]:
-                        if data[meta["name"]] is None:
+                        if data.get(meta["name"]) is None:
                             data[meta["name"]] = self.get_default_value(meta["type"])
 
             return metric_layer_result
@@ -984,15 +989,15 @@ def get_snql_query(self) -> Request:
         and returns one or more equivalent snql query(ies).
         """
 
-        if self.use_metrics_layer or self._on_demand_spec:
+        if self.use_metrics_layer or self.on_demand_metrics_enabled:
             from sentry.snuba.metrics import SnubaQueryBuilder
             from sentry.snuba.metrics.mqb_query_transformer import (
                 transform_mqb_query_to_metrics_query,
             )
 
             snuba_request = self.get_metrics_layer_snql_query()
 
-            if self._on_demand_spec:
+            if self.on_demand_metrics_enabled:
                 metrics_query = self._get_on_demand_metrics_query()
             else:
                 metrics_query = transform_mqb_query_to_metrics_query(
@@ -1068,6 +1073,7 @@ def resolve_split_granularity(self) -> Tuple[List[Condition], Optional[Granulari
 
 class TimeseriesMetricQueryBuilder(MetricsQueryBuilder):
     time_alias = "time"
+    is_timeseries_query = True
 
     def __init__(
         self,
@@ -1081,6 +1087,7 @@ def __init__(
         limit: Optional[int] = 10000,
         use_metrics_layer: Optional[bool] = False,
         groupby: Optional[Column] = None,
+        on_demand_metrics_enabled: Optional[bool] = False,
     ):
         super().__init__(
             params=params,
@@ -1091,6 +1098,7 @@ def __init__(
             auto_fields=False,
             functions_acl=functions_acl,
             use_metrics_layer=use_metrics_layer,
+            on_demand_metrics_enabled=on_demand_metrics_enabled,
         )
         if self.granularity.granularity > interval:
             for granularity in constants.METRICS_GRANULARITIES:
@@ -1210,7 +1218,7 @@ def get_snql_query(self) -> List[Request]:
         return queries
 
     def run_query(self, referrer: str, use_cache: bool = False) -> Any:
-        if self.use_metrics_layer:
+        if self.use_metrics_layer or self.on_demand_metrics_enabled:
             from sentry.snuba.metrics.datasource import get_series
             from sentry.snuba.metrics.mqb_query_transformer import (
                 transform_mqb_query_to_metrics_query,
@@ -1219,13 +1227,16 @@ def run_query(self, referrer: str, use_cache: bool = False) -> Any:
             snuba_query = self.get_snql_query()[0].query
             try:
                 with sentry_sdk.start_span(op="metric_layer", description="transform_query"):
-                    metric_query = transform_mqb_query_to_metrics_query(
-                        snuba_query, self.is_alerts_query
-                    )
+                    if self.on_demand_metrics_enabled:
+                        metrics_query = self._get_on_demand_metrics_query()
+                    else:
+                        metrics_query = transform_mqb_query_to_metrics_query(
+                            snuba_query, self.is_alerts_query
+                        )
                 with sentry_sdk.start_span(op="metric_layer", description="run_query"):
                     metrics_data = get_series(
                         projects=self.params.projects,
-                        metrics_query=metric_query,
+                        metrics_query=metrics_query,
                         use_case_id=UseCaseID.TRANSACTIONS
                         if self.is_performance
                         else UseCaseID.SESSIONS,

src/sentry/snuba/discover.py

@@ -262,6 +262,7 @@ def timeseries_query(
     allow_metric_aggregates=False,
     has_metrics=False,
     use_metrics_layer=False,
+    on_demand_metrics_enabled=False,
 ):
     """
     High-level API for doing arbitrary user timeseries queries against events.

src/sentry/snuba/entity_subscription.py

@@ -305,6 +305,10 @@ def __init__(
         self.use_metrics_layer = features.has(
             "organizations:use-metrics-layer", Organization.objects.get_from_cache(id=self.org_id)
         )
+        self.on_demand_metrics_enabled = features.has(
+            "organizations:on-demand-metrics-extraction",
+            Organization.objects.get_from_cache(id=self.org_id),
+        )
 
     @abstractmethod
     def get_snql_aggregations(self) -> List[str]:
@@ -378,6 +382,7 @@ def build_query_builder(
             skip_time_conditions=True,
             granularity=self.get_granularity(),
             use_metrics_layer=self.use_metrics_layer,
+            on_demand_metrics_enabled=self.on_demand_metrics_enabled,
         )
         extra_conditions = self.get_snql_extra_conditions()
 

src/sentry/snuba/functions.py

@@ -37,6 +37,7 @@ def query(
     has_metrics: bool = False,
     functions_acl: Optional[List[str]] = None,
     use_metrics_layer: bool = False,
+    on_demand_metrics_enabled: bool = False,
 ) -> Any:
     if not selected_columns:
         raise InvalidSearchQuery("No columns selected")
@@ -73,6 +74,7 @@ def timeseries_query(
     allow_metric_aggregates: bool = False,
     has_metrics: bool = False,
     use_metrics_layer: bool = False,
+    on_demand_metrics_enabled: bool = False,
 ) -> Any:
     builder = ProfileFunctionsTimeseriesQueryBuilder(
         dataset=Dataset.Functions,

src/sentry/snuba/issue_platform.py

@@ -109,6 +109,7 @@ def timeseries_query(
     allow_metric_aggregates=False,
     has_metrics=False,
     use_metrics_layer=False,
+    on_demand_metrics_enabled=False,
 ):
     """
     High-level API for doing arbitrary user timeseries queries against events.

src/sentry/snuba/metrics/extraction.py

@@ -121,7 +121,8 @@
     "p99": "d",
 }
 
-# Query fields that on their own do not require on-demand metric extraction.
+# Query fields that on their own do not require on-demand metric extraction but if present in an on-demand query
+# will be converted to metric extraction conditions.
 _STANDARD_METRIC_FIELDS = [
     "release",
     "dist",
@@ -135,7 +136,9 @@
     "browser.name",
     "os.name",
     "geo.country_code",
+    # These fields are skipped during on demand spec generation and will not be converted to metric extraction conditions
     "event.type",
+    "project",
 ]
 
 # Operators used in ``ComparingRuleCondition``.
@@ -222,9 +225,8 @@ def is_on_demand_query(
     for field in _get_aggregate_fields(aggregate):
         if not _is_standard_metrics_field(field):
             return True
-
     try:
-        return not _is_standard_metrics_query(event_search.parse_search_query(query or ""))
+        return not _is_standard_metrics_query(event_search.parse_search_query(query))
     except InvalidSearchQuery:
         logger.error(f"Failed to parse search query: {query}", exc_info=True)
         return False
@@ -304,11 +306,18 @@ def __init__(self, field: str, query: str):
 
         """
 
-        # On-demand metrics are implicitly transaction metrics. Remove the
-        # filter from the query since it can't be translated to a RuleCondition.
-        self._query = re.sub(r"event\.type:transaction\s*", "", query)
+        self._init__query(query)
         self._init_aggregate(field)
 
+    def _init__query(self, query: str) -> None:
+        # On-demand metrics are implicitly transaction metrics. Remove the
+        # filters from the query that can't be translated to a RuleCondition.
+        query = re.sub(r"event\.type:transaction\s*", "", query)
+        # extend the following to also support project:"some-project"
+        query = re.sub(r"project:[\w\"]+\s*", "", query)
+
+        self._query = query.strip()
+
     def _init_aggregate(self, aggregate: str) -> None:
         """
         Extracts the field name, metric type and metric operation from a Discover

src/sentry/snuba/metrics_enhanced_performance.py

@@ -31,6 +31,7 @@ def query(
     transform_alias_to_input_format=False,
     has_metrics: bool = True,
     use_metrics_layer: bool = False,
+    on_demand_metrics_enabled: bool = False,
 ):
     metrics_compatible = not equations
 
@@ -105,6 +106,7 @@ def timeseries_query(
     functions_acl: Optional[List[str]] = None,
     has_metrics: bool = True,
     use_metrics_layer: bool = False,
+    on_demand_metrics_enabled: bool = False,
 ) -> SnubaTSResult:
     """
     High-level API for doing arbitrary user timeseries queries against events.