feat(flamegraphs): Support functions flamegraphs for continuous profi… (#80822)

…ling

Previously, when generating flamegraphs from functions, it was only
capable of using transaction profiles. This adds support for generating
flamegraphs for a function using continuous profiles.

Author: Zylphrex

src/sentry/profiles/flamegraph.py

@@ -295,7 +295,7 @@ class ProfilerMeta:
     thread_id: str
     start: float
     end: float
-    transaction_id: str
+    transaction_id: str | None = None
 
     def as_condition(self) -> Condition:
         return And(
@@ -341,14 +341,13 @@ def get_profile_candidates(self) -> ProfileCandidates:
         raise NotImplementedError
 
     def get_profile_candidates_from_functions(self) -> ProfileCandidates:
-        # TODO: continuous profiles support
         max_profiles = options.get("profiling.flamegraph.profile-set.size")
 
         builder = ProfileFunctionsQueryBuilder(
             dataset=Dataset.Functions,
             params={},
             snuba_params=self.snuba_params,
-            selected_columns=["project.id", "timestamp", "unique_examples()"],
+            selected_columns=["project.id", "timestamp", "all_examples()"],
             query=self.query,
             limit=max_profiles,
             config=QueryBuilderConfig(
@@ -365,22 +364,34 @@ def get_profile_candidates_from_functions(self) -> ProfileCandidates:
         results = builder.process_results(results)
 
         transaction_profile_candidates: list[TransactionProfileCandidate] = []
+        profiler_metas: list[ProfilerMeta] = []
 
         for row in results["data"]:
             project = row["project.id"]
-            for example in row["unique_examples()"]:
+            for example in row["all_examples()"]:
                 if len(transaction_profile_candidates) > max_profiles:
                     break
-                transaction_profile_candidates.append(
-                    {
-                        "project_id": project,
-                        "profile_id": example,
-                    }
-                )
+                if "profile_id" in example:
+                    transaction_profile_candidates.append(
+                        {
+                            "project_id": project,
+                            "profile_id": example["profile_id"],
+                        }
+                    )
+                elif "profiler_id" in example:
+                    profiler_metas.append(
+                        ProfilerMeta(
+                            project_id=project,
+                            profiler_id=example["profiler_id"],
+                            thread_id=example["thread_id"],
+                            start=example["start"],
+                            end=example["end"],
+                        )
+                    )
 
         return {
             "transaction": transaction_profile_candidates,
-            "continuous": [],
+            "continuous": self.get_chunks_for_profilers(profiler_metas),
         }
 
     def get_profile_candidates_from_transactions(self) -> ProfileCandidates:
@@ -499,17 +510,19 @@ def get_chunks_for_profilers(
                     if start > profiler_meta.end or end < profiler_meta.start:
                         continue
 
-                    continuous_profile_candidates.append(
-                        {
-                            "project_id": profiler_meta.project_id,
-                            "profiler_id": profiler_meta.profiler_id,
-                            "chunk_id": row["chunk_id"],
-                            "thread_id": profiler_meta.thread_id,
-                            "start": str(int(profiler_meta.start * 1.0e9)),
-                            "end": str(int(profiler_meta.end * 1.0e9)),
-                            "transaction_id": profiler_meta.transaction_id,
-                        }
-                    )
+                    candidate: ContinuousProfileCandidate = {
+                        "project_id": profiler_meta.project_id,
+                        "profiler_id": profiler_meta.profiler_id,
+                        "chunk_id": row["chunk_id"],
+                        "thread_id": profiler_meta.thread_id,
+                        "start": str(int(profiler_meta.start * 1e9)),
+                        "end": str(int(profiler_meta.end * 1e9)),
+                    }
+
+                    if profiler_meta.transaction_id is not None:
+                        candidate["transaction_id"] = profiler_meta.transaction_id
+
+                    continuous_profile_candidates.append(candidate)
 
         return continuous_profile_candidates
 

tests/sentry/api/endpoints/test_organization_profiling_profiles.py

@@ -121,6 +121,7 @@ def setUp(self):
         self.login_as(user=self.user)
         self.url = reverse(self.endpoint, args=(self.organization.slug,))
         self.ten_mins_ago = before_now(minutes=10)
+        self.hour_ago = before_now(hours=1).replace(minute=0, second=0, microsecond=0)
 
     def do_request(self, query, features=None, compat=True, **kwargs):
         if features is None:
@@ -404,40 +405,65 @@ def test_queries_profile_candidates_from_profiles(self):
 
             assert profiles_snql_request.dataset == Dataset.Profiles.value
 
+    @patch("sentry.profiles.flamegraph.bulk_snuba_queries", wraps=bulk_snuba_queries)
     @patch("sentry.search.events.builder.base.raw_snql_query", wraps=raw_snql_query)
     @patch("sentry.api.endpoints.organization_profiling_profiles.proxy_profiling_service")
     def test_queries_profile_candidates_from_functions_with_data(
         self,
         mock_proxy_profiling_service,
         mock_raw_snql_query,
+        mock_bulk_snuba_queries,
     ):
-        data = load_data("transaction", timestamp=self.ten_mins_ago)
+        data = load_data("transaction", timestamp=self.hour_ago)
         data["transaction"] = "foo"
         profile_id = uuid4().hex
         data.setdefault("contexts", {}).setdefault("profile", {})["profile_id"] = profile_id
 
-        stored = self.store_functions(
+        stored_1 = self.store_functions(
             [
                 {
-                    "self_times_ns": [100],
+                    "self_times_ns": [100_000_000],
                     "package": "foo",
                     "function": "bar",
                     "in_app": True,
                 },
             ],
             project=self.project,
             transaction=data,
+            timestamp=self.hour_ago,
         )
+        stored_2 = self.store_functions_chunk(
+            [
+                {
+                    "self_times_ns": [100_000_000],
+                    "package": "foo",
+                    "function": "bar",
+                    "thread_id": "1",
+                    "in_app": True,
+                },
+            ],
+            project=self.project,
+            timestamp=self.hour_ago,
+        )
+
+        chunk = {
+            "project_id": self.project.id,
+            "profiler_id": stored_2["profiler_id"],
+            "chunk_id": stored_2["chunk_id"],
+            "start_timestamp": self.hour_ago.isoformat(),
+            "end_timestamp": (self.hour_ago + timedelta(microseconds=100_000)).isoformat(),
+        }
+
+        mock_bulk_snuba_queries.return_value = [{"data": [chunk]}]
 
         mock_proxy_profiling_service.return_value = HttpResponse(status=200)
 
-        fingerprint = stored["functions"][0]["fingerprint"]
+        fingerprint = stored_1["functions"][0]["fingerprint"]
 
         response = self.do_request(
             {
                 "project": [self.project.id],
                 "dataSource": "functions",
-                "query": "transaction:foo",
                 "fingerprint": str(fingerprint),
             },
         )
@@ -457,7 +483,6 @@ def test_queries_profile_candidates_from_functions_with_data(
             )
             in snql_request.query.where
         )
-        assert Condition(Column("transaction_name"), Op.EQ, "foo") in snql_request.query.where
 
         mock_proxy_profiling_service.assert_called_once_with(
             method="POST",
@@ -469,7 +494,18 @@ def test_queries_profile_candidates_from_functions_with_data(
                         "profile_id": profile_id,
                     },
                 ],
-                "continuous": [],
+                "continuous": [
+                    {
+                        "project_id": self.project.id,
+                        "profiler_id": stored_2["profiler_id"],
+                        "chunk_id": stored_2["chunk_id"],
+                        "thread_id": "1",
+                        "start": str(int(self.hour_ago.timestamp() * 1e9)),
+                        "end": str(
+                            int((self.hour_ago + timedelta(microseconds=100_000)).timestamp() * 1e9)
+                        ),
+                    },
+                ],
             },
         )