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,
@@ -217,6 +218,9 @@ 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=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(query)
  except InvalidSearchQuery:
  logger.error(f"Failed to parse search query: {query}", exc_info=True)
  return False
@@ -250,6 +252,10 @@ def _get_aggregate_fields(aggregate: str) -> Sequence[str]:
  return []
 
 
+def is_standard_metrics_query(query: Optional[str] = "") -> bool:
+ return _is_standard_metrics_query(event_search.parse_search_query(query))
+
+
 def _is_standard_metrics_query(tokens: Sequence[QueryToken]) -> bool:
  """
  Recursively checks if any of the supplied token contain search filters that can't be handled by standard metrics.
@@ -304,11 +310,17 @@ 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)
+ 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.