ref(releasehealth): Implement get_project_releases_by_stability on me…

…trics backend [INGEST-249] (#29352)

get_project_releases_by_stability

src/sentry/api/endpoints/organization_releases.py

@@ -40,7 +40,7 @@
 )
 from sentry.search.events.filter import handle_operator_negation, parse_semver
 from sentry.signals import release_created
-from sentry.snuba.sessions import STATS_PERIODS, get_project_releases_by_stability
+from sentry.snuba.sessions import STATS_PERIODS
 from sentry.utils.cache import cache
 from sentry.utils.compat import zip as izip
 from sentry.utils.sdk import bind_organization_context, configure_scope
@@ -344,7 +344,7 @@ def qs_load_func(queryset, total_offset, qs_offset, limit):
 
             paginator_cls = MergingOffsetPaginator
             paginator_kwargs.update(
-                data_load_func=lambda offset, limit: get_project_releases_by_stability(
+                data_load_func=lambda offset, limit: release_health.get_project_releases_by_stability(
                     project_ids=filter_params["project_id"],
                     environments=filter_params.get("environment"),
                     scope=sort,

src/sentry/release_health/base.py

@@ -209,6 +209,7 @@ class ReleaseHealthBackend(Service):  # type: ignore
         "get_project_release_stats",
         "get_project_sessions_count",
         "get_num_sessions_per_project",
+        "get_project_releases_by_stability",
     )
 
     def get_current_and_previous_crash_free_rates(
@@ -430,3 +431,17 @@ def get_num_sessions_per_project(
         Additionally
         """
         raise NotImplementedError()
+
+    def get_project_releases_by_stability(
+        self,
+        project_ids: Sequence[ProjectId],
+        offset: Optional[int],
+        limit: Optional[int],
+        scope: str,
+        stats_period: Optional[str] = None,
+        environments: Optional[Sequence[str]] = None,
+    ) -> Sequence[ProjectRelease]:
+        """Given some project IDs returns adoption rates that should be updated
+        on the postgres tables.
+        """
+        raise NotImplementedError()

src/sentry/release_health/metrics.py

@@ -20,8 +20,8 @@
 )
 
 import pytz
-from snuba_sdk import Column, Condition, Entity, Function, Op, Query
-from snuba_sdk.expressions import Expression, Granularity
+from snuba_sdk import Column, Condition, Direction, Entity, Function, Op, OrderBy, Query
+from snuba_sdk.expressions import Expression, Granularity, Limit, Offset
 from snuba_sdk.query import SelectableExpression
 
 from sentry.models import Environment
@@ -1762,3 +1762,166 @@ def get_num_sessions_per_project(
         )["data"]
 
         return [(row["project_id"], row["value"]) for row in rows]
+
+    def get_project_releases_by_stability(
+        self,
+        project_ids: Sequence[ProjectId],
+        offset: Optional[int],
+        limit: Optional[int],
+        scope: str,
+        stats_period: Optional[str] = None,
+        environments: Optional[Sequence[str]] = None,
+    ) -> Sequence[ProjectRelease]:
+
+        if len(project_ids) == 0:
+            return []
+
+        org_id = self._get_org_id(project_ids)
+        environments_ids: Optional[Sequence[int]] = None
+
+        if environments is not None:
+            environments_ids = get_tag_values_list(org_id, environments)
+            if len(environments_ids) == 0:
+                return []
+
+        release_column_name = tag_key(org_id, "release")
+
+        if stats_period is None:
+            stats_period = "24h"
+
+        # Special rule that we support sorting by the last 24h only.
+        if scope.endswith("_24h"):
+            scope = scope[:-4]
+            stats_period = "24h"
+
+        now = datetime.now(pytz.utc)
+        granularity, stats_start, _ = get_rollup_starts_and_buckets(stats_period)
+
+        query_cols = [
+            Column("project_id"),
+            Column(release_column_name),
+        ]
+        group_by = [
+            Column("project_id"),
+            Column(release_column_name),
+        ]
+
+        where_clause = [
+            Condition(Column("org_id"), Op.EQ, org_id),
+            Condition(Column("project_id"), Op.IN, project_ids),
+            Condition(Column("timestamp"), Op.GTE, stats_start),
+            Condition(Column("timestamp"), Op.LT, now),
+        ]
+
+        if environments_ids is not None:
+            environment_column_name = tag_key(org_id, "environment")
+            where_clause.append(Condition(Column(environment_column_name), Op.IN, environments_ids))
+
+        having_clause: Optional[List[Condition]] = None
+
+        status_init = tag_value(org_id, "init")
+        status_crashed = tag_value(org_id, "crashed")
+        session_status_column_name = tag_key(org_id, "session.status")
+
+        order_by_clause = None
+        if scope == "crash_free_sessions":
+            order_by_clause = [
+                OrderBy(
+                    exp=Function(
+                        "divide",
+                        parameters=[
+                            Function(
+                                "sumIf",
+                                parameters=[
+                                    Column("value"),
+                                    Function(
+                                        "equals",
+                                        [Column(session_status_column_name), status_crashed],
+                                    ),
+                                ],
+                            ),
+                            Function(
+                                "sumIf",
+                                parameters=[
+                                    Column("value"),
+                                    Function(
+                                        "equals", [Column(session_status_column_name), status_init]
+                                    ),
+                                ],
+                            ),
+                        ],
+                    ),
+                    direction=Direction.DESC,
+                )
+            ]
+            where_clause.append(Condition(Column("metric_id"), Op.EQ, metric_id(org_id, "session")))
+            entity = Entity(EntityKey.MetricsCounters.value)
+        elif scope == "sessions":
+            order_by_clause = [
+                OrderBy(exp=Function("sum", [Column("value")], "value"), direction=Direction.DESC)
+            ]
+            where_clause.append(Condition(Column("metric_id"), Op.EQ, metric_id(org_id, "session")))
+            entity = Entity(EntityKey.MetricsCounters.value)
+        elif scope == "crash_free_users":
+            order_by_clause = [
+                OrderBy(
+                    exp=Function(
+                        "divide",
+                        parameters=[
+                            Function(
+                                "uniqIf",
+                                parameters=[
+                                    Column("value"),
+                                    Function(
+                                        "equals",
+                                        [Column(session_status_column_name), status_crashed],
+                                    ),
+                                ],
+                            ),
+                            Function(
+                                "uniqIf",
+                                parameters=[
+                                    Column("value"),
+                                    Function(
+                                        "equals", [Column(session_status_column_name), status_init]
+                                    ),
+                                ],
+                            ),
+                        ],
+                    ),
+                    direction=Direction.DESC,
+                )
+            ]
+            where_clause.append(Condition(Column("metric_id"), Op.EQ, metric_id(org_id, "user")))
+            entity = Entity(EntityKey.MetricsSets.value)
+            having_clause = [Condition(Function("uniq", [Column("value")], "users"), Op.GT, 0)]
+        else:  # users
+            users_column = Function("uniq", [Column("value")], "users")
+            order_by_clause = [OrderBy(exp=users_column, direction=Direction.DESC)]
+            where_clause.append(Condition(Column("metric_id"), Op.EQ, metric_id(org_id, "user")))
+            entity = Entity(EntityKey.MetricsSets.value)
+            having_clause = [Condition(users_column, Op.GT, 0)]
+
+        query = Query(
+            dataset=Dataset.Metrics.value,
+            match=entity,
+            select=query_cols,
+            where=where_clause,
+            having=having_clause,
+            orderby=order_by_clause,
+            groupby=group_by,
+            offset=Offset(offset) if offset is not None else None,
+            limit=Limit(limit) if limit is not None else None,
+            granularity=Granularity(granularity),
+        )
+
+        rows = raw_snql_query(
+            query,
+            referrer="release_health.metrics.get_project_releases_by_stability",
+            use_cache=False,
+        )
+
+        def extract_row_info(row: Mapping[str, Union[OrganizationId, str]]) -> ProjectRelease:
+            return row.get("project_id"), reverse_tag_value(org_id, row.get(release_column_name))  # type: ignore
+
+        return [extract_row_info(row) for row in rows["data"]]

src/sentry/release_health/sessions.py

@@ -31,6 +31,7 @@
     _get_num_sessions_per_project,
     _get_oldest_health_data_for_releases,
     _get_project_release_stats,
+    _get_project_releases_by_stability,
     _get_project_releases_count,
     _get_project_sessions_count,
     _get_release_adoption,
@@ -224,3 +225,16 @@ def get_num_sessions_per_project(
         return _get_num_sessions_per_project(  # type: ignore
             project_ids, start, end, environment_ids, rollup
         )
+
+    def get_project_releases_by_stability(
+        self,
+        project_ids: Sequence[ProjectId],
+        offset: Optional[int],
+        limit: Optional[int],
+        scope: str,
+        stats_period: Optional[str] = None,
+        environments: Optional[Sequence[str]] = None,
+    ) -> Sequence[ProjectRelease]:
+        return _get_project_releases_by_stability(  # type: ignore
+            project_ids, offset, limit, scope, stats_period, environments
+        )

src/sentry/snuba/sessions.py

@@ -149,7 +149,7 @@ def _check_releases_have_health_data(
     return {row["release"] for row in data}
 
 
-def get_project_releases_by_stability(
+def _get_project_releases_by_stability(
     project_ids, offset, limit, scope, stats_period=None, environments=None
 ):
     """Given some project IDs returns adoption rates that should be updated

tests/snuba/sessions/test_sessions.py

@@ -8,7 +8,7 @@
 from sentry.release_health.base import OverviewStat
 from sentry.release_health.metrics import MetricsReleaseHealthBackend
 from sentry.release_health.sessions import SessionsReleaseHealthBackend
-from sentry.snuba.sessions import _make_stats, get_project_releases_by_stability
+from sentry.snuba.sessions import _make_stats
 from sentry.testutils import SnubaTestCase, TestCase
 from sentry.testutils.cases import SessionMetricsTestCase
 from sentry.utils.dates import to_timestamp
@@ -57,10 +57,14 @@ def setUp(self):
         self.session_started = time.time() // 60 * 60
         self.session_release = "foo@1.0.0"
         self.session_crashed_release = "foo@2.0.0"
+        session_1 = "5d52fd05-fcc9-4bf3-9dc9-267783670341"
+        session_2 = "5e910c1a-6941-460e-9843-24103fb6a63c"
+        session_3 = "a148c0c5-06a2-423b-8901-6b43b812cf82"
+        user_1 = "39887d89-13b2-4c84-8c23-5d13d2102666"
         self.store_session(
             {
-                "session_id": "5d52fd05-fcc9-4bf3-9dc9-267783670341",
-                "distinct_id": "39887d89-13b2-4c84-8c23-5d13d2102666",
+                "session_id": session_1,
+                "distinct_id": user_1,
                 "status": "exited",
                 "seq": 0,
                 "release": self.session_release,
@@ -77,8 +81,8 @@ def setUp(self):
 
         self.store_session(
             {
-                "session_id": "5e910c1a-6941-460e-9843-24103fb6a63c",
-                "distinct_id": "39887d89-13b2-4c84-8c23-5d13d2102666",
+                "session_id": session_2,
+                "distinct_id": user_1,
                 "status": "ok",
                 "seq": 0,
                 "release": self.session_release,
@@ -95,8 +99,8 @@ def setUp(self):
 
         self.store_session(
             {
-                "session_id": "5e910c1a-6941-460e-9843-24103fb6a63c",
-                "distinct_id": "39887d89-13b2-4c84-8c23-5d13d2102666",
+                "session_id": session_2,
+                "distinct_id": user_1,
                 "status": "exited",
                 "seq": 1,
                 "release": self.session_release,
@@ -113,8 +117,8 @@ def setUp(self):
 
         self.store_session(
             {
-                "session_id": "a148c0c5-06a2-423b-8901-6b43b812cf82",
-                "distinct_id": "39887d89-13b2-4c84-8c23-5d13d2102666",
+                "session_id": session_3,
+                "distinct_id": user_1,
                 "status": "crashed",
                 "seq": 0,
                 "release": self.session_crashed_release,
@@ -223,7 +227,7 @@ def test_get_project_releases_by_stability(self):
         )
 
         for scope in "sessions", "users":
-            data = get_project_releases_by_stability(
+            data = self.backend.get_project_releases_by_stability(
                 [self.project.id], offset=0, limit=100, scope=scope, stats_period="24h"
             )
 
@@ -237,8 +241,28 @@ def test_get_project_releases_by_stability_for_crash_free_sort(self):
         Test that ensures that using crash free rate sort options, returns a list of ASC releases
         according to the chosen crash_free sort option
         """
+
+        # add another user to session_release to make sure that they are sorted correctly
+        self.store_session(
+            {
+                "session_id": str(uuid.uuid4()),
+                "distinct_id": str(uuid.uuid4()),
+                "status": "exited",
+                "seq": 0,
+                "release": self.session_release,
+                "environment": "prod",
+                "retention_days": 90,
+                "org_id": self.project.organization_id,
+                "project_id": self.project.id,
+                "duration": 60.0,
+                "errors": 0,
+                "started": self.session_started,
+                "received": self.received,
+            }
+        )
+
         for scope in "crash_free_sessions", "crash_free_users":
-            data = get_project_releases_by_stability(
+            data = self.backend.get_project_releases_by_stability(
                 [self.project.id], offset=0, limit=100, scope=scope, stats_period="24h"
             )
             assert data == [
@@ -267,15 +291,15 @@ def test_get_project_releases_by_stability_for_releases_with_users_data(self):
                 "received": self.received,
             }
         )
-        data = get_project_releases_by_stability(
+        data = self.backend.get_project_releases_by_stability(
             [self.project.id], offset=0, limit=100, scope="users", stats_period="24h"
         )
         assert set(data) == {
             (self.project.id, self.session_release),
             (self.project.id, self.session_crashed_release),
         }
 
-        data = get_project_releases_by_stability(
+        data = self.backend.get_project_releases_by_stability(
             [self.project.id], offset=0, limit=100, scope="crash_free_users", stats_period="24h"
         )
         assert set(data) == {