We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support profiling and escalating issues in group counts
Counts for profiling (not GAed) and performance issues are not in the events dataset (where errors go).

In order to support escalation for these type of issues we would need to look at the search issues dataset

Basically `if category == ERROR: <Access errors dataset> else <access search issues dataset>`

Change [this query](https://github.com/getsentry/sentry/blob/master/src/sentry/issues/escalating.py#L51-L83) to look at the Dataset.IssuePlatform dataset and the EntityKey.IssuePlatform entity when we’re not looking at errors.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/sentry/issues/escalating.py`
Content:
```
1 """This module has the logic for querying Snuba for the hourly event count for a list of groups.
2 This is later used for generating group forecasts for determining when a group may be escalating.
3 """
4 
5 import logging
6 from collections import defaultdict
7 from datetime import datetime, timedelta
8 from typing import Dict, List, Sequence, Tuple, TypedDict
9 
10 from snuba_sdk import (
11     Column,
12     Condition,
13     Direction,
14     Entity,
15     Function,
16     Limit,
17     Offset,
18     Op,
19     OrderBy,
20     Query,
21     Request,
22 )
23 
24 from sentry import analytics
25 from sentry.issues.escalating_group_forecast import EscalatingGroupForecast
26 from sentry.issues.escalating_issues_alg import GroupCount
27 from sentry.issues.grouptype import GroupCategory
28 from sentry.models import Group
29 from sentry.models.group import GroupStatus
30 from sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox
31 from sentry.snuba.dataset import Dataset, EntityKey
32 from sentry.types.group import GroupSubStatus
33 from sentry.utils.cache import cache
34 from sentry.utils.snuba import raw_snql_query
35 
36 logger = logging.getLogger(__name__)
37 
38 __all__ = ["query_groups_past_counts", "parse_groups_past_counts"]
39 
40 REFERRER = "sentry.issues.escalating"
41 ELEMENTS_PER_SNUBA_PAGE = 10000  # This is the maximum value for Snuba
42 # The amount of data needed to generate a group forecast
43 BUCKETS_PER_GROUP = 7 * 24
44 ONE_WEEK_DURATION = 7
45 IS_ESCALATING_REFERRER = "sentry.issues.escalating.is_escalating"
46 GROUP_HOURLY_COUNT_TTL = 60
47 
48 GroupsCountResponse = TypedDict(
49     "GroupsCountResponse",
50     {"group_id": int, "hourBucket": str, "count()": int, "project_id": int},
51 )
52 
53 ParsedGroupsCount = Dict[int, GroupCount]
54 
55 
56 def query_groups_past_counts(groups: Sequence[Group]) -> List[GroupsCountResponse]:
57     """Query Snuba for the counts for every group bucketed into hours.
58 
59     It optimizes the query by guaranteeing that we look at group_ids that are from the same project id.
60     This is important for Snuba as the data is stored in blocks related to the project id.
61 
62     We maximize the number of projects and groups to reduce the total number of Snuba queries.
63     Each project may not have enough groups in order to reach the max number of returned
64     elements (ELEMENTS_PER_SNUBA_PAGE), thus, projects with few groups should be grouped together until
65     we get at least a certain number of groups.
66 
67     NOTE: Groups with less than the maximum number of buckets (think of groups with just 1 event or less
68     than 7 days old) will skew the optimization since we may only get one page and less elements than the max
69     ELEMENTS_PER_SNUBA_PAGE.
70     """
71     all_results = []  # type: ignore[var-annotated]
72     if not groups:
73         return all_results
74 
75     start_date, end_date = _start_and_end_dates()
76     group_ids_by_project = _extract_project_and_group_ids(groups)
77     proj_ids, group_ids = [], []
78     processed_projects = 0
79     total_projects_count = len(group_ids_by_project)
80     organization_id = groups[0].project.organization.id
81 
82     # This iteration guarantees that all groups for a project will be queried in the same call
83     # and only one page where the groups could be mixed with groups from another project
84     # Iterating over the sorted keys guarantees results for tests
85     for proj_id in sorted(group_ids_by_project.keys()):
86         _group_ids = group_ids_by_project[proj_id]
87         # Add them to the list of projects and groups to query
88         proj_ids.append(proj_id)
89         group_ids += _group_ids
90         processed_projects += 1
91         potential_num_elements = len(_group_ids) * BUCKETS_PER_GROUP
92         # This is trying to maximize the number of groups on the first page
93         if (
94             processed_projects < total_projects_count
95             and potential_num_elements < ELEMENTS_PER_SNUBA_PAGE
96         ):
97             continue
98 
99         # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids
100         all_results += _query_with_pagination(
101             organization_id, proj_ids, group_ids, start_date, end_date
102         )
103         # We're ready for a new set of projects and ids
104         proj_ids, group_ids = [], []
105 
106     return all_results
107 
108 
109 def _query_with_pagination(
110     organization_id: int,
111     project_ids: Sequence[int],
112     group_ids: Sequence[int],
113     start_date: datetime,
114     end_date: datetime,
115 ) -> List[GroupsCountResponse]:
116     """Query Snuba for event counts for the given list of project ids and groups ids in
117     a time range."""
118     all_results = []
119     offset = 0
120     while True:
121         query = _generate_query(project_ids, group_ids, offset, start_date, end_date)
122         request = Request(
123             dataset=_issue_category_dataset(GroupCategory.ERROR),
124             app_id=REFERRER,
125             query=query,
126             tenant_ids={"referrer": REFERRER, "organization_id": organization_id},
127         )
128         results = raw_snql_query(request, referrer=REFERRER)["data"]
129         all_results += results
130         offset += ELEMENTS_PER_SNUBA_PAGE
131         if not results or len(results) < ELEMENTS_PER_SNUBA_PAGE:
132             break
133 
134     return all_results
135 
136 
137 def _generate_query(
138     project_ids: Sequence[int],
139     group_ids: Sequence[int],
140     offset: int,
141     start_date: datetime,
142     end_date: datetime,
143 ) -> Query:
144     """This simply generates a query based on the passed parameters"""
145     group_id_col = Column("group_id")
146     proj_id_col = Column("project_id")
147     return Query(
148         match=Entity(_issue_category_entity(GroupCategory.ERROR)),
149         select=[
150             proj_id_col,
151             group_id_col,
152             Function("toStartOfHour", [Column("timestamp")], "hourBucket"),
153             Function("count", []),
154         ],
155         groupby=[proj_id_col, group_id_col, Column("hourBucket")],
156         where=[
157             Condition(proj_id_col, Op.IN, Function("tuple", project_ids)),
158             Condition(Column("group_id"), Op.IN, Function("tuple", group_ids)),
159             Condition(Column("timestamp"), Op.GTE, start_date),
160             Condition(Column("timestamp"), Op.LT, end_date),
161         ],
162         limit=Limit(ELEMENTS_PER_SNUBA_PAGE),
163         offset=Offset(offset),
164         orderby=[
165             OrderBy(proj_id_col, Direction.ASC),
166             OrderBy(group_id_col, Direction.ASC),
167             OrderBy(Column("hourBucket"), Direction.ASC),
168         ],
169     )
170 
171 
172 def _start_and_end_dates(hours: int = BUCKETS_PER_GROUP) -> Tuple[datetime, datetime]:
173     """Return the start and end date of N hours time range."""
174     end_datetime = datetime.now()
175     return end_datetime - timedelta(hours=hours), end_datetime
176 
177 
178 def _extract_project_and_group_ids(groups: Sequence[Group]) -> Dict[int, List[int]]:
179     """Return all project and group IDs from a list of Group"""
180     group_ids_by_project: Dict[int, List[int]] = defaultdict(list)
181     for group in groups:
182         group_ids_by_project[group.project_id].append(group.id)
183 
184     return group_ids_by_project
185 
186 
187 def get_group_hourly_count(group: Group) -> int:
188     """Return the number of events a group has had today in the last hour"""
189     key = f"hourly-group-count:{group.project.id}:{group.id}"
190     hourly_count = cache.get(key)
191 
192     if hourly_count is None:
193         now = datetime.now()
194         current_hour = now.replace(minute=0, second=0, microsecond=0)
195         query = Query(
196             match=Entity(_issue_category_entity(group.issue_category)),
197             select=[
198                 Function("count", []),
199             ],
200             where=[
201                 Condition(Column("project_id"), Op.EQ, group.project.id),
202                 Condition(Column("group_id"), Op.EQ, group.id),
203                 Condition(Column("timestamp"), Op.GTE, current_hour),
204                 Condition(Column("timestamp"), Op.LT, now),
205             ],
206         )
207         request = Request(
208             dataset=_issue_category_dataset(group.issue_category),
209             app_id=IS_ESCALATING_REFERRER,
210             query=query,
211             tenant_ids={
212                 "referrer": IS_ESCALATING_REFERRER,
213                 "organization_id": group.project.organization.id,
214             },
215         )
216         hourly_count = int(
217             raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)["data"][0]["count()"]
218         )
219         cache.set(key, hourly_count, GROUP_HOURLY_COUNT_TTL)
220     return int(hourly_count)
221 
222 
223 def is_escalating(group: Group) -> bool:
224     """Return boolean depending on if the group is escalating or not"""
225     group_hourly_count = get_group_hourly_count(group)
226     forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)
227     # Check if current event occurance is greater than forecast for today's date
228     if group_hourly_count > forecast_today:
229         group.substatus = GroupSubStatus.ESCALATING
230         group.status = GroupStatus.UNRESOLVED
231         add_group_to_inbox(group, GroupInboxReason.ESCALATING)
232 
233         analytics.record(
234             "issue.escalating",
235             organization_id=group.project.organization.id,
236             project_id=group.project.id,
237             group_id=group.id,
238         )
239         return True
240     return False
241 
242 
243 def parse_groups_past_counts(response: Sequence[GroupsCountResponse]) -> ParsedGroupsCount:
244     """
245     Return the parsed snuba response for groups past counts to be used in generate_issue_forecast.
246     ParsedGroupCount is of the form {<group_id>: {"intervals": [str], "data": [int]}}.
247 
248     `response`: Snuba response for group event counts
249     """
250     group_counts: ParsedGroupsCount = {}
251     group_ids_list = group_counts.keys()
252     for data in response:
253         group_id = data["group_id"]
254         if group_id not in group_ids_list:
255             group_counts[group_id] = {
256                 "intervals": [data["hourBucket"]],
257                 "data": [data["count()"]],
258             }
259         else:
260             group_counts[group_id]["intervals"].append(data["hourBucket"])
261             group_counts[group_id]["data"].append(data["count()"])
262     return group_counts
263 
264 
265 def _issue_category_dataset(category: GroupCategory) -> Dataset:
266     if category == GroupCategory.ERROR:
267         return Dataset.Events.value
268     else:
269         raise NotImplementedError
270 
271 
272 def _issue_category_entity(category: GroupCategory) -> EntityKey:
273     if category == GroupCategory.ERROR:
274         return EntityKey.Events.value
275     else:
276         raise NotImplementedError
277
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/sentry/issues/escalating.py b/src/sentry/issues/escalating.py
--- a/src/sentry/issues/escalating.py
+++ b/src/sentry/issues/escalating.py
@@ -5,7 +5,7 @@
 import logging
 from collections import defaultdict
 from datetime import datetime, timedelta
-from typing import Dict, List, Sequence, Tuple, TypedDict
+from typing import Dict, List, Optional, Sequence, Tuple, TypedDict
 
 from snuba_sdk import (
     Column,
@@ -73,6 +73,34 @@
         return all_results
 
     start_date, end_date = _start_and_end_dates()
+
+    # Error groups use the events dataset while profile and perf groups use the issue platform dataset
+    error_groups: List[Group] = []
+    other_groups: List[Group] = []
+    for g in groups:
+        if g.issue_category == GroupCategory.ERROR:
+            error_groups.append(g)
+        else:
+            other_groups.append(g)
+
+    all_results += _process_groups(error_groups, start_date, end_date, GroupCategory.ERROR)
+    all_results += _process_groups(other_groups, start_date, end_date)
+
+    return all_results
+
+
+def _process_groups(
+    groups: Sequence[Group],
+    start_date: datetime,
+    end_date: datetime,
+    category: Optional[GroupCategory] = None,
+) -> List[GroupsCountResponse]:
+    """Given a list of groups, query Snuba for their hourly bucket count.
+    The category defines which Snuba dataset and entity we query."""
+    all_results = []  # type: ignore[var-annotated]
+    if not groups:
+        return all_results
+
     group_ids_by_project = _extract_project_and_group_ids(groups)
     proj_ids, group_ids = [], []
     processed_projects = 0
@@ -98,7 +126,7 @@
 
         # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids
         all_results += _query_with_pagination(
-            organization_id, proj_ids, group_ids, start_date, end_date
+            organization_id, proj_ids, group_ids, start_date, end_date, category
         )
         # We're ready for a new set of projects and ids
         proj_ids, group_ids = [], []
@@ -112,15 +140,16 @@
     group_ids: Sequence[int],
     start_date: datetime,
     end_date: datetime,
+    category: Optional[GroupCategory],
 ) -> List[GroupsCountResponse]:
     """Query Snuba for event counts for the given list of project ids and groups ids in
     a time range."""
     all_results = []
     offset = 0
     while True:
-        query = _generate_query(project_ids, group_ids, offset, start_date, end_date)
+        query = _generate_query(project_ids, group_ids, offset, start_date, end_date, category)
         request = Request(
-            dataset=_issue_category_dataset(GroupCategory.ERROR),
+            dataset=_issue_category_dataset(category),
             app_id=REFERRER,
             query=query,
             tenant_ids={"referrer": REFERRER, "organization_id": organization_id},
@@ -140,12 +169,13 @@
     offset: int,
     start_date: datetime,
     end_date: datetime,
+    category: Optional[GroupCategory],
 ) -> Query:
     """This simply generates a query based on the passed parameters"""
     group_id_col = Column("group_id")
     proj_id_col = Column("project_id")
     return Query(
-        match=Entity(_issue_category_entity(GroupCategory.ERROR)),
+        match=Entity(_issue_category_entity(category)),
         select=[
             proj_id_col,
             group_id_col,
@@ -262,15 +292,11 @@
     return group_counts
 
 
-def _issue_category_dataset(category: GroupCategory) -> Dataset:
-    if category == GroupCategory.ERROR:
-        return Dataset.Events.value
-    else:
-        raise NotImplementedError
+def _issue_category_dataset(category: Optional[GroupCategory]) -> Dataset:
+    return Dataset.Events.value if category == GroupCategory.ERROR else Dataset.IssuePlatform.value
 
 
-def _issue_category_entity(category: GroupCategory) -> EntityKey:
-    if category == GroupCategory.ERROR:
-        return EntityKey.Events.value
-    else:
-        raise NotImplementedError
+def _issue_category_entity(category: Optional[GroupCategory]) -> EntityKey:
+    return (
+        EntityKey.Events.value if category == GroupCategory.ERROR else EntityKey.IssuePlatform.value
+    )

{"golden_diff": "diff --git a/src/sentry/issues/escalating.py b/src/sentry/issues/escalating.py\n--- a/src/sentry/issues/escalating.py\n+++ b/src/sentry/issues/escalating.py\n@@ -5,7 +5,7 @@\n import logging\n from collections import defaultdict\n from datetime import datetime, timedelta\n-from typing import Dict, List, Sequence, Tuple, TypedDict\n+from typing import Dict, List, Optional, Sequence, Tuple, TypedDict\n \n from snuba_sdk import (\n     Column,\n@@ -73,6 +73,34 @@\n         return all_results\n \n     start_date, end_date = _start_and_end_dates()\n+\n+    # Error groups use the events dataset while profile and perf groups use the issue platform dataset\n+    error_groups: List[Group] = []\n+    other_groups: List[Group] = []\n+    for g in groups:\n+        if g.issue_category == GroupCategory.ERROR:\n+            error_groups.append(g)\n+        else:\n+            other_groups.append(g)\n+\n+    all_results += _process_groups(error_groups, start_date, end_date, GroupCategory.ERROR)\n+    all_results += _process_groups(other_groups, start_date, end_date)\n+\n+    return all_results\n+\n+\n+def _process_groups(\n+    groups: Sequence[Group],\n+    start_date: datetime,\n+    end_date: datetime,\n+    category: Optional[GroupCategory] = None,\n+) -> List[GroupsCountResponse]:\n+    \"\"\"Given a list of groups, query Snuba for their hourly bucket count.\n+    The category defines which Snuba dataset and entity we query.\"\"\"\n+    all_results = []  # type: ignore[var-annotated]\n+    if not groups:\n+        return all_results\n+\n     group_ids_by_project = _extract_project_and_group_ids(groups)\n     proj_ids, group_ids = [], []\n     processed_projects = 0\n@@ -98,7 +126,7 @@\n \n         # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids\n         all_results += _query_with_pagination(\n-            organization_id, proj_ids, group_ids, start_date, end_date\n+            organization_id, proj_ids, group_ids, start_date, end_date, category\n         )\n         # We're ready for a new set of projects and ids\n         proj_ids, group_ids = [], []\n@@ -112,15 +140,16 @@\n     group_ids: Sequence[int],\n     start_date: datetime,\n     end_date: datetime,\n+    category: Optional[GroupCategory],\n ) -> List[GroupsCountResponse]:\n     \"\"\"Query Snuba for event counts for the given list of project ids and groups ids in\n     a time range.\"\"\"\n     all_results = []\n     offset = 0\n     while True:\n-        query = _generate_query(project_ids, group_ids, offset, start_date, end_date)\n+        query = _generate_query(project_ids, group_ids, offset, start_date, end_date, category)\n         request = Request(\n-            dataset=_issue_category_dataset(GroupCategory.ERROR),\n+            dataset=_issue_category_dataset(category),\n             app_id=REFERRER,\n             query=query,\n             tenant_ids={\"referrer\": REFERRER, \"organization_id\": organization_id},\n@@ -140,12 +169,13 @@\n     offset: int,\n     start_date: datetime,\n     end_date: datetime,\n+    category: Optional[GroupCategory],\n ) -> Query:\n     \"\"\"This simply generates a query based on the passed parameters\"\"\"\n     group_id_col = Column(\"group_id\")\n     proj_id_col = Column(\"project_id\")\n     return Query(\n-        match=Entity(_issue_category_entity(GroupCategory.ERROR)),\n+        match=Entity(_issue_category_entity(category)),\n         select=[\n             proj_id_col,\n             group_id_col,\n@@ -262,15 +292,11 @@\n     return group_counts\n \n \n-def _issue_category_dataset(category: GroupCategory) -> Dataset:\n-    if category == GroupCategory.ERROR:\n-        return Dataset.Events.value\n-    else:\n-        raise NotImplementedError\n+def _issue_category_dataset(category: Optional[GroupCategory]) -> Dataset:\n+    return Dataset.Events.value if category == GroupCategory.ERROR else Dataset.IssuePlatform.value\n \n \n-def _issue_category_entity(category: GroupCategory) -> EntityKey:\n-    if category == GroupCategory.ERROR:\n-        return EntityKey.Events.value\n-    else:\n-        raise NotImplementedError\n+def _issue_category_entity(category: Optional[GroupCategory]) -> EntityKey:\n+    return (\n+        EntityKey.Events.value if category == GroupCategory.ERROR else EntityKey.IssuePlatform.value\n+    )\n", "issue": "Support profiling and escalating issues in group counts\nCounts for profiling (not GAed) and performance issues are not in the events dataset (where errors go).\r\n\r\nIn order to support escalation for these type of issues we would need to look at the search issues dataset\r\n\r\nBasically `if category == ERROR: <Access errors dataset> else <access search issues dataset>`\r\n\r\nChange [this query](https://github.com/getsentry/sentry/blob/master/src/sentry/issues/escalating.py#L51-L83) to look at the Dataset.IssuePlatform dataset and the EntityKey.IssuePlatform entity when we\u2019re not looking at errors.\n", "before_files": [{"content": "\"\"\"This module has the logic for querying Snuba for the hourly event count for a list of groups.\nThis is later used for generating group forecasts for determining when a group may be escalating.\n\"\"\"\n\nimport logging\nfrom collections import defaultdict\nfrom datetime import datetime, timedelta\nfrom typing import Dict, List, Sequence, Tuple, TypedDict\n\nfrom snuba_sdk import (\n    Column,\n    Condition,\n    Direction,\n    Entity,\n    Function,\n    Limit,\n    Offset,\n    Op,\n    OrderBy,\n    Query,\n    Request,\n)\n\nfrom sentry import analytics\nfrom sentry.issues.escalating_group_forecast import EscalatingGroupForecast\nfrom sentry.issues.escalating_issues_alg import GroupCount\nfrom sentry.issues.grouptype import GroupCategory\nfrom sentry.models import Group\nfrom sentry.models.group import GroupStatus\nfrom sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox\nfrom sentry.snuba.dataset import Dataset, EntityKey\nfrom sentry.types.group import GroupSubStatus\nfrom sentry.utils.cache import cache\nfrom sentry.utils.snuba import raw_snql_query\n\nlogger = logging.getLogger(__name__)\n\n__all__ = [\"query_groups_past_counts\", \"parse_groups_past_counts\"]\n\nREFERRER = \"sentry.issues.escalating\"\nELEMENTS_PER_SNUBA_PAGE = 10000  # This is the maximum value for Snuba\n# The amount of data needed to generate a group forecast\nBUCKETS_PER_GROUP = 7 * 24\nONE_WEEK_DURATION = 7\nIS_ESCALATING_REFERRER = \"sentry.issues.escalating.is_escalating\"\nGROUP_HOURLY_COUNT_TTL = 60\n\nGroupsCountResponse = TypedDict(\n    \"GroupsCountResponse\",\n    {\"group_id\": int, \"hourBucket\": str, \"count()\": int, \"project_id\": int},\n)\n\nParsedGroupsCount = Dict[int, GroupCount]\n\n\ndef query_groups_past_counts(groups: Sequence[Group]) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for the counts for every group bucketed into hours.\n\n    It optimizes the query by guaranteeing that we look at group_ids that are from the same project id.\n    This is important for Snuba as the data is stored in blocks related to the project id.\n\n    We maximize the number of projects and groups to reduce the total number of Snuba queries.\n    Each project may not have enough groups in order to reach the max number of returned\n    elements (ELEMENTS_PER_SNUBA_PAGE), thus, projects with few groups should be grouped together until\n    we get at least a certain number of groups.\n\n    NOTE: Groups with less than the maximum number of buckets (think of groups with just 1 event or less\n    than 7 days old) will skew the optimization since we may only get one page and less elements than the max\n    ELEMENTS_PER_SNUBA_PAGE.\n    \"\"\"\n    all_results = []  # type: ignore[var-annotated]\n    if not groups:\n        return all_results\n\n    start_date, end_date = _start_and_end_dates()\n    group_ids_by_project = _extract_project_and_group_ids(groups)\n    proj_ids, group_ids = [], []\n    processed_projects = 0\n    total_projects_count = len(group_ids_by_project)\n    organization_id = groups[0].project.organization.id\n\n    # This iteration guarantees that all groups for a project will be queried in the same call\n    # and only one page where the groups could be mixed with groups from another project\n    # Iterating over the sorted keys guarantees results for tests\n    for proj_id in sorted(group_ids_by_project.keys()):\n        _group_ids = group_ids_by_project[proj_id]\n        # Add them to the list of projects and groups to query\n        proj_ids.append(proj_id)\n        group_ids += _group_ids\n        processed_projects += 1\n        potential_num_elements = len(_group_ids) * BUCKETS_PER_GROUP\n        # This is trying to maximize the number of groups on the first page\n        if (\n            processed_projects < total_projects_count\n            and potential_num_elements < ELEMENTS_PER_SNUBA_PAGE\n        ):\n            continue\n\n        # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids\n        all_results += _query_with_pagination(\n            organization_id, proj_ids, group_ids, start_date, end_date\n        )\n        # We're ready for a new set of projects and ids\n        proj_ids, group_ids = [], []\n\n    return all_results\n\n\ndef _query_with_pagination(\n    organization_id: int,\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    start_date: datetime,\n    end_date: datetime,\n) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for event counts for the given list of project ids and groups ids in\n    a time range.\"\"\"\n    all_results = []\n    offset = 0\n    while True:\n        query = _generate_query(project_ids, group_ids, offset, start_date, end_date)\n        request = Request(\n            dataset=_issue_category_dataset(GroupCategory.ERROR),\n            app_id=REFERRER,\n            query=query,\n            tenant_ids={\"referrer\": REFERRER, \"organization_id\": organization_id},\n        )\n        results = raw_snql_query(request, referrer=REFERRER)[\"data\"]\n        all_results += results\n        offset += ELEMENTS_PER_SNUBA_PAGE\n        if not results or len(results) < ELEMENTS_PER_SNUBA_PAGE:\n            break\n\n    return all_results\n\n\ndef _generate_query(\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    offset: int,\n    start_date: datetime,\n    end_date: datetime,\n) -> Query:\n    \"\"\"This simply generates a query based on the passed parameters\"\"\"\n    group_id_col = Column(\"group_id\")\n    proj_id_col = Column(\"project_id\")\n    return Query(\n        match=Entity(_issue_category_entity(GroupCategory.ERROR)),\n        select=[\n            proj_id_col,\n            group_id_col,\n            Function(\"toStartOfHour\", [Column(\"timestamp\")], \"hourBucket\"),\n            Function(\"count\", []),\n        ],\n        groupby=[proj_id_col, group_id_col, Column(\"hourBucket\")],\n        where=[\n            Condition(proj_id_col, Op.IN, Function(\"tuple\", project_ids)),\n            Condition(Column(\"group_id\"), Op.IN, Function(\"tuple\", group_ids)),\n            Condition(Column(\"timestamp\"), Op.GTE, start_date),\n            Condition(Column(\"timestamp\"), Op.LT, end_date),\n        ],\n        limit=Limit(ELEMENTS_PER_SNUBA_PAGE),\n        offset=Offset(offset),\n        orderby=[\n            OrderBy(proj_id_col, Direction.ASC),\n            OrderBy(group_id_col, Direction.ASC),\n            OrderBy(Column(\"hourBucket\"), Direction.ASC),\n        ],\n    )\n\n\ndef _start_and_end_dates(hours: int = BUCKETS_PER_GROUP) -> Tuple[datetime, datetime]:\n    \"\"\"Return the start and end date of N hours time range.\"\"\"\n    end_datetime = datetime.now()\n    return end_datetime - timedelta(hours=hours), end_datetime\n\n\ndef _extract_project_and_group_ids(groups: Sequence[Group]) -> Dict[int, List[int]]:\n    \"\"\"Return all project and group IDs from a list of Group\"\"\"\n    group_ids_by_project: Dict[int, List[int]] = defaultdict(list)\n    for group in groups:\n        group_ids_by_project[group.project_id].append(group.id)\n\n    return group_ids_by_project\n\n\ndef get_group_hourly_count(group: Group) -> int:\n    \"\"\"Return the number of events a group has had today in the last hour\"\"\"\n    key = f\"hourly-group-count:{group.project.id}:{group.id}\"\n    hourly_count = cache.get(key)\n\n    if hourly_count is None:\n        now = datetime.now()\n        current_hour = now.replace(minute=0, second=0, microsecond=0)\n        query = Query(\n            match=Entity(_issue_category_entity(group.issue_category)),\n            select=[\n                Function(\"count\", []),\n            ],\n            where=[\n                Condition(Column(\"project_id\"), Op.EQ, group.project.id),\n                Condition(Column(\"group_id\"), Op.EQ, group.id),\n                Condition(Column(\"timestamp\"), Op.GTE, current_hour),\n                Condition(Column(\"timestamp\"), Op.LT, now),\n            ],\n        )\n        request = Request(\n            dataset=_issue_category_dataset(group.issue_category),\n            app_id=IS_ESCALATING_REFERRER,\n            query=query,\n            tenant_ids={\n                \"referrer\": IS_ESCALATING_REFERRER,\n                \"organization_id\": group.project.organization.id,\n            },\n        )\n        hourly_count = int(\n            raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)[\"data\"][0][\"count()\"]\n        )\n        cache.set(key, hourly_count, GROUP_HOURLY_COUNT_TTL)\n    return int(hourly_count)\n\n\ndef is_escalating(group: Group) -> bool:\n    \"\"\"Return boolean depending on if the group is escalating or not\"\"\"\n    group_hourly_count = get_group_hourly_count(group)\n    forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)\n    # Check if current event occurance is greater than forecast for today's date\n    if group_hourly_count > forecast_today:\n        group.substatus = GroupSubStatus.ESCALATING\n        group.status = GroupStatus.UNRESOLVED\n        add_group_to_inbox(group, GroupInboxReason.ESCALATING)\n\n        analytics.record(\n            \"issue.escalating\",\n            organization_id=group.project.organization.id,\n            project_id=group.project.id,\n            group_id=group.id,\n        )\n        return True\n    return False\n\n\ndef parse_groups_past_counts(response: Sequence[GroupsCountResponse]) -> ParsedGroupsCount:\n    \"\"\"\n    Return the parsed snuba response for groups past counts to be used in generate_issue_forecast.\n    ParsedGroupCount is of the form {<group_id>: {\"intervals\": [str], \"data\": [int]}}.\n\n    `response`: Snuba response for group event counts\n    \"\"\"\n    group_counts: ParsedGroupsCount = {}\n    group_ids_list = group_counts.keys()\n    for data in response:\n        group_id = data[\"group_id\"]\n        if group_id not in group_ids_list:\n            group_counts[group_id] = {\n                \"intervals\": [data[\"hourBucket\"]],\n                \"data\": [data[\"count()\"]],\n            }\n        else:\n            group_counts[group_id][\"intervals\"].append(data[\"hourBucket\"])\n            group_counts[group_id][\"data\"].append(data[\"count()\"])\n    return group_counts\n\n\ndef _issue_category_dataset(category: GroupCategory) -> Dataset:\n    if category == GroupCategory.ERROR:\n        return Dataset.Events.value\n    else:\n        raise NotImplementedError\n\n\ndef _issue_category_entity(category: GroupCategory) -> EntityKey:\n    if category == GroupCategory.ERROR:\n        return EntityKey.Events.value\n    else:\n        raise NotImplementedError\n", "path": "src/sentry/issues/escalating.py"}], "after_files": [{"content": "\"\"\"This module has the logic for querying Snuba for the hourly event count for a list of groups.\nThis is later used for generating group forecasts for determining when a group may be escalating.\n\"\"\"\n\nimport logging\nfrom collections import defaultdict\nfrom datetime import datetime, timedelta\nfrom typing import Dict, List, Optional, Sequence, Tuple, TypedDict\n\nfrom snuba_sdk import (\n    Column,\n    Condition,\n    Direction,\n    Entity,\n    Function,\n    Limit,\n    Offset,\n    Op,\n    OrderBy,\n    Query,\n    Request,\n)\n\nfrom sentry import analytics\nfrom sentry.issues.escalating_group_forecast import EscalatingGroupForecast\nfrom sentry.issues.escalating_issues_alg import GroupCount\nfrom sentry.issues.grouptype import GroupCategory\nfrom sentry.models import Group\nfrom sentry.models.group import GroupStatus\nfrom sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox\nfrom sentry.snuba.dataset import Dataset, EntityKey\nfrom sentry.types.group import GroupSubStatus\nfrom sentry.utils.cache import cache\nfrom sentry.utils.snuba import raw_snql_query\n\nlogger = logging.getLogger(__name__)\n\n__all__ = [\"query_groups_past_counts\", \"parse_groups_past_counts\"]\n\nREFERRER = \"sentry.issues.escalating\"\nELEMENTS_PER_SNUBA_PAGE = 10000  # This is the maximum value for Snuba\n# The amount of data needed to generate a group forecast\nBUCKETS_PER_GROUP = 7 * 24\nONE_WEEK_DURATION = 7\nIS_ESCALATING_REFERRER = \"sentry.issues.escalating.is_escalating\"\nGROUP_HOURLY_COUNT_TTL = 60\n\nGroupsCountResponse = TypedDict(\n    \"GroupsCountResponse\",\n    {\"group_id\": int, \"hourBucket\": str, \"count()\": int, \"project_id\": int},\n)\n\nParsedGroupsCount = Dict[int, GroupCount]\n\n\ndef query_groups_past_counts(groups: Sequence[Group]) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for the counts for every group bucketed into hours.\n\n    It optimizes the query by guaranteeing that we look at group_ids that are from the same project id.\n    This is important for Snuba as the data is stored in blocks related to the project id.\n\n    We maximize the number of projects and groups to reduce the total number of Snuba queries.\n    Each project may not have enough groups in order to reach the max number of returned\n    elements (ELEMENTS_PER_SNUBA_PAGE), thus, projects with few groups should be grouped together until\n    we get at least a certain number of groups.\n\n    NOTE: Groups with less than the maximum number of buckets (think of groups with just 1 event or less\n    than 7 days old) will skew the optimization since we may only get one page and less elements than the max\n    ELEMENTS_PER_SNUBA_PAGE.\n    \"\"\"\n    all_results = []  # type: ignore[var-annotated]\n    if not groups:\n        return all_results\n\n    start_date, end_date = _start_and_end_dates()\n\n    # Error groups use the events dataset while profile and perf groups use the issue platform dataset\n    error_groups: List[Group] = []\n    other_groups: List[Group] = []\n    for g in groups:\n        if g.issue_category == GroupCategory.ERROR:\n            error_groups.append(g)\n        else:\n            other_groups.append(g)\n\n    all_results += _process_groups(error_groups, start_date, end_date, GroupCategory.ERROR)\n    all_results += _process_groups(other_groups, start_date, end_date)\n\n    return all_results\n\n\ndef _process_groups(\n    groups: Sequence[Group],\n    start_date: datetime,\n    end_date: datetime,\n    category: Optional[GroupCategory] = None,\n) -> List[GroupsCountResponse]:\n    \"\"\"Given a list of groups, query Snuba for their hourly bucket count.\n    The category defines which Snuba dataset and entity we query.\"\"\"\n    all_results = []  # type: ignore[var-annotated]\n    if not groups:\n        return all_results\n\n    group_ids_by_project = _extract_project_and_group_ids(groups)\n    proj_ids, group_ids = [], []\n    processed_projects = 0\n    total_projects_count = len(group_ids_by_project)\n    organization_id = groups[0].project.organization.id\n\n    # This iteration guarantees that all groups for a project will be queried in the same call\n    # and only one page where the groups could be mixed with groups from another project\n    # Iterating over the sorted keys guarantees results for tests\n    for proj_id in sorted(group_ids_by_project.keys()):\n        _group_ids = group_ids_by_project[proj_id]\n        # Add them to the list of projects and groups to query\n        proj_ids.append(proj_id)\n        group_ids += _group_ids\n        processed_projects += 1\n        potential_num_elements = len(_group_ids) * BUCKETS_PER_GROUP\n        # This is trying to maximize the number of groups on the first page\n        if (\n            processed_projects < total_projects_count\n            and potential_num_elements < ELEMENTS_PER_SNUBA_PAGE\n        ):\n            continue\n\n        # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids\n        all_results += _query_with_pagination(\n            organization_id, proj_ids, group_ids, start_date, end_date, category\n        )\n        # We're ready for a new set of projects and ids\n        proj_ids, group_ids = [], []\n\n    return all_results\n\n\ndef _query_with_pagination(\n    organization_id: int,\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    start_date: datetime,\n    end_date: datetime,\n    category: Optional[GroupCategory],\n) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for event counts for the given list of project ids and groups ids in\n    a time range.\"\"\"\n    all_results = []\n    offset = 0\n    while True:\n        query = _generate_query(project_ids, group_ids, offset, start_date, end_date, category)\n        request = Request(\n            dataset=_issue_category_dataset(category),\n            app_id=REFERRER,\n            query=query,\n            tenant_ids={\"referrer\": REFERRER, \"organization_id\": organization_id},\n        )\n        results = raw_snql_query(request, referrer=REFERRER)[\"data\"]\n        all_results += results\n        offset += ELEMENTS_PER_SNUBA_PAGE\n        if not results or len(results) < ELEMENTS_PER_SNUBA_PAGE:\n            break\n\n    return all_results\n\n\ndef _generate_query(\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    offset: int,\n    start_date: datetime,\n    end_date: datetime,\n    category: Optional[GroupCategory],\n) -> Query:\n    \"\"\"This simply generates a query based on the passed parameters\"\"\"\n    group_id_col = Column(\"group_id\")\n    proj_id_col = Column(\"project_id\")\n    return Query(\n        match=Entity(_issue_category_entity(category)),\n        select=[\n            proj_id_col,\n            group_id_col,\n            Function(\"toStartOfHour\", [Column(\"timestamp\")], \"hourBucket\"),\n            Function(\"count\", []),\n        ],\n        groupby=[proj_id_col, group_id_col, Column(\"hourBucket\")],\n        where=[\n            Condition(proj_id_col, Op.IN, Function(\"tuple\", project_ids)),\n            Condition(Column(\"group_id\"), Op.IN, Function(\"tuple\", group_ids)),\n            Condition(Column(\"timestamp\"), Op.GTE, start_date),\n            Condition(Column(\"timestamp\"), Op.LT, end_date),\n        ],\n        limit=Limit(ELEMENTS_PER_SNUBA_PAGE),\n        offset=Offset(offset),\n        orderby=[\n            OrderBy(proj_id_col, Direction.ASC),\n            OrderBy(group_id_col, Direction.ASC),\n            OrderBy(Column(\"hourBucket\"), Direction.ASC),\n        ],\n    )\n\n\ndef _start_and_end_dates(hours: int = BUCKETS_PER_GROUP) -> Tuple[datetime, datetime]:\n    \"\"\"Return the start and end date of N hours time range.\"\"\"\n    end_datetime = datetime.now()\n    return end_datetime - timedelta(hours=hours), end_datetime\n\n\ndef _extract_project_and_group_ids(groups: Sequence[Group]) -> Dict[int, List[int]]:\n    \"\"\"Return all project and group IDs from a list of Group\"\"\"\n    group_ids_by_project: Dict[int, List[int]] = defaultdict(list)\n    for group in groups:\n        group_ids_by_project[group.project_id].append(group.id)\n\n    return group_ids_by_project\n\n\ndef get_group_hourly_count(group: Group) -> int:\n    \"\"\"Return the number of events a group has had today in the last hour\"\"\"\n    key = f\"hourly-group-count:{group.project.id}:{group.id}\"\n    hourly_count = cache.get(key)\n\n    if hourly_count is None:\n        now = datetime.now()\n        current_hour = now.replace(minute=0, second=0, microsecond=0)\n        query = Query(\n            match=Entity(_issue_category_entity(group.issue_category)),\n            select=[\n                Function(\"count\", []),\n            ],\n            where=[\n                Condition(Column(\"project_id\"), Op.EQ, group.project.id),\n                Condition(Column(\"group_id\"), Op.EQ, group.id),\n                Condition(Column(\"timestamp\"), Op.GTE, current_hour),\n                Condition(Column(\"timestamp\"), Op.LT, now),\n            ],\n        )\n        request = Request(\n            dataset=_issue_category_dataset(group.issue_category),\n            app_id=IS_ESCALATING_REFERRER,\n            query=query,\n            tenant_ids={\n                \"referrer\": IS_ESCALATING_REFERRER,\n                \"organization_id\": group.project.organization.id,\n            },\n        )\n        hourly_count = int(\n            raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)[\"data\"][0][\"count()\"]\n        )\n        cache.set(key, hourly_count, GROUP_HOURLY_COUNT_TTL)\n    return int(hourly_count)\n\n\ndef is_escalating(group: Group) -> bool:\n    \"\"\"Return boolean depending on if the group is escalating or not\"\"\"\n    group_hourly_count = get_group_hourly_count(group)\n    forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)\n    # Check if current event occurance is greater than forecast for today's date\n    if group_hourly_count > forecast_today:\n        group.substatus = GroupSubStatus.ESCALATING\n        group.status = GroupStatus.UNRESOLVED\n        add_group_to_inbox(group, GroupInboxReason.ESCALATING)\n\n        analytics.record(\n            \"issue.escalating\",\n            organization_id=group.project.organization.id,\n            project_id=group.project.id,\n            group_id=group.id,\n        )\n        return True\n    return False\n\n\ndef parse_groups_past_counts(response: Sequence[GroupsCountResponse]) -> ParsedGroupsCount:\n    \"\"\"\n    Return the parsed snuba response for groups past counts to be used in generate_issue_forecast.\n    ParsedGroupCount is of the form {<group_id>: {\"intervals\": [str], \"data\": [int]}}.\n\n    `response`: Snuba response for group event counts\n    \"\"\"\n    group_counts: ParsedGroupsCount = {}\n    group_ids_list = group_counts.keys()\n    for data in response:\n        group_id = data[\"group_id\"]\n        if group_id not in group_ids_list:\n            group_counts[group_id] = {\n                \"intervals\": [data[\"hourBucket\"]],\n                \"data\": [data[\"count()\"]],\n            }\n        else:\n            group_counts[group_id][\"intervals\"].append(data[\"hourBucket\"])\n            group_counts[group_id][\"data\"].append(data[\"count()\"])\n    return group_counts\n\n\ndef _issue_category_dataset(category: Optional[GroupCategory]) -> Dataset:\n    return Dataset.Events.value if category == GroupCategory.ERROR else Dataset.IssuePlatform.value\n\n\ndef _issue_category_entity(category: Optional[GroupCategory]) -> EntityKey:\n    return (\n        EntityKey.Events.value if category == GroupCategory.ERROR else EntityKey.IssuePlatform.value\n    )\n", "path": "src/sentry/issues/escalating.py"}]}