ref(statistical-detectors): Unify breakpoint detection interface

src/sentry/api/endpoints/organization_events_trends_v2.py

@@ -7,12 +7,10 @@
 from typing import Any, Dict, List, cast
 
 import sentry_sdk
-from django.conf import settings
 from rest_framework.exceptions import ParseError
 from rest_framework.request import Request
 from rest_framework.response import Response
 from snuba_sdk import Column
-from urllib3 import Retry
 
 from sentry import features
 from sentry.api.api_publish_status import ApiPublishStatus
@@ -22,14 +20,14 @@
 from sentry.issues.grouptype import PerformanceDurationRegressionGroupType
 from sentry.issues.issue_occurrence import IssueEvidence, IssueOccurrence
 from sentry.issues.producer import produce_occurrence_to_kafka
-from sentry.net.http import connection_from_url
 from sentry.search.events.constants import METRICS_GRANULARITIES
+from sentry.seer.utils import detect_breakpoints
 from sentry.snuba import metrics_performance
 from sentry.snuba.discover import create_result_key, zerofill
 from sentry.snuba.metrics_performance import query as metrics_query
 from sentry.snuba.referrer import Referrer
 from sentry.types.ratelimit import RateLimit, RateLimitCategory
-from sentry.utils import json, metrics
+from sentry.utils import metrics
 from sentry.utils.snuba import SnubaTSResult
 
 logger = logging.getLogger(__name__)
@@ -51,28 +49,9 @@
 DEFAULT_CONCURRENT_RATE_LIMIT = 15
 ORGANIZATION_RATE_LIMIT = 30
 
-ads_connection_pool = connection_from_url(
- settings.ANOMALY_DETECTION_URL,
- retries=Retry(
- total=5,
- status_forcelist=[408, 429, 502, 503, 504],
- ),
- timeout=settings.ANOMALY_DETECTION_TIMEOUT,
-)
-
 _query_thread_pool = ThreadPoolExecutor()
 
 
-def get_trends(snuba_io):
- response = ads_connection_pool.urlopen(
- "POST",
- "/trends/breakpoint-detector",
- body=json.dumps(snuba_io),
- headers={"content-type": "application/json;charset=utf-8"},
- )
- return json.loads(response.data)
-
-
 @region_silo_endpoint
 class OrganizationEventsNewTrendsStatsEndpoint(OrganizationEventsV2EndpointBase):
  publish_status = {
@@ -296,7 +275,7 @@ def get_trends_data(stats_data, request):
  trends_requests.append(trends_request)
 
  # send the data to microservice
- results = list(_query_thread_pool.map(get_trends, trends_requests))
+ results = list(_query_thread_pool.map(detect_breakpoints, trends_requests))
  trend_results = []
 
  # append all the results

src/sentry/api/endpoints/organization_profiling_functions.py

@@ -4,38 +4,26 @@
 from enum import Enum
 from typing import Any
 
-from django.conf import settings
 from rest_framework import serializers
 from rest_framework.request import Request
 from rest_framework.response import Response
-from urllib3 import Retry
 
 from sentry import features
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import region_silo_endpoint
 from sentry.api.bases import NoProjects, OrganizationEventsV2EndpointBase
 from sentry.api.paginator import GenericOffsetPaginator
 from sentry.exceptions import InvalidSearchQuery
-from sentry.net.http import connection_from_url
 from sentry.search.events.builder import ProfileTopFunctionsTimeseriesQueryBuilder
 from sentry.search.events.types import QueryBuilderConfig
+from sentry.seer.utils import detect_breakpoints
 from sentry.snuba import functions
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.referrer import Referrer
-from sentry.utils import json
 from sentry.utils.dates import parse_stats_period, validate_interval
 from sentry.utils.sdk import set_measurement
 from sentry.utils.snuba import bulk_snql_query
 
-ads_connection_pool = connection_from_url(
- settings.ANOMALY_DETECTION_URL,
- retries=Retry(
- total=5,
- status_forcelist=[408, 429, 502, 503, 504],
- ),
- timeout=settings.ANOMALY_DETECTION_TIMEOUT,
-)
-
 TOP_FUNCTIONS_LIMIT = 50
 FUNCTIONS_PER_QUERY = 10
 
@@ -192,7 +180,7 @@ def get_trends_data(stats_data):
  "trendFunction": data["function"],
  }
 
- return trends_query(trends_request)
+ return detect_breakpoints(trends_request)
 
  stats_data = self.get_event_stats_data(
  request,
@@ -320,14 +308,3 @@ def get_interval_from_range(date_range: timedelta) -> str:
  return "2h"
 
  return "1h"
-
-
-def trends_query(trends_request):
- response = ads_connection_pool.urlopen(
- "POST",
- "/trends/breakpoint-detector",
- body=json.dumps(trends_request),
- headers={"content-type": "application/json;charset=utf-8"},
- )
-
- return json.loads(response.data)["data"]

src/sentry/seer/utils.py

@@ -0,0 +1,24 @@
+from django.conf import settings
+from urllib3 import Retry
+
+from sentry.net.http import connection_from_url
+from sentry.utils import json
+
+seer_connection_pool = connection_from_url(
+ settings.ANOMALY_DETECTION_URL,
+ retries=Retry(
+ total=5,
+ status_forcelist=[408, 429, 502, 503, 504],
+ ),
+ timeout=settings.ANOMALY_DETECTION_TIMEOUT,
+)
+
+
+def detect_breakpoints(breakpoint_request):
+ response = seer_connection_pool.urlopen(
+ "POST",
+ "/trends/breakpoint-detector",
+ body=json.dumps(breakpoint_request),
+ headers={"content-type": "application/json;charset=utf-8"},
+ )
+ return json.loads(response.data)

tests/sentry/api/endpoints/test_organization_events_trends_v2.py

@@ -91,8 +91,8 @@ def test_no_project(self):
  assert response.status_code == 200, response.content
  assert response.data == []
 
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
- def test_simple_with_trends(self, mock_get_trends):
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
+ def test_simple_with_trends(self, mock_detect_breakpoints):
  mock_trends_result = [
  {
  "project": self.project.slug,
@@ -102,7 +102,7 @@ def test_simple_with_trends(self, mock_get_trends):
  "trend_percentage": 0.88,
  }
  ]
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(
@@ -129,10 +129,10 @@ def test_simple_with_trends(self, mock_get_trends):
  assert len(result_stats) > 0
  assert len(result_stats.get(f"{self.project.slug},foo", [])) > 0
 
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
- def test_simple_with_no_trends(self, mock_get_trends):
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
+ def test_simple_with_no_trends(self, mock_detect_breakpoints):
  mock_trends_result: List[Union[Dict[str, Any], None]] = []
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(
@@ -158,10 +158,10 @@ def test_simple_with_no_trends(self, mock_get_trends):
 
  assert len(result_stats) == 0
 
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
- def test_simple_with_transaction_query(self, mock_get_trends):
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
+ def test_simple_with_transaction_query(self, mock_detect_breakpoints):
  mock_trends_result: List[Union[Dict[str, Any], None]] = []
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  self.store_performance_metric(
  name=TransactionMRI.DURATION.value,
@@ -186,14 +186,14 @@ def test_simple_with_transaction_query(self, mock_get_trends):
  },
  )
 
- trends_call_args_data = mock_get_trends.call_args[0][0]["data"]
+ trends_call_args_data = mock_detect_breakpoints.call_args[0][0]["data"]
  assert len(trends_call_args_data.get(f"{self.project.slug},foo")) > 0
  assert len(trends_call_args_data.get(f"{self.project.slug},bar", [])) == 0
 
  assert response.status_code == 200, response.content
 
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
- def test_simple_with_trends_p75(self, mock_get_trends):
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
+ def test_simple_with_trends_p75(self, mock_detect_breakpoints):
  mock_trends_result = [
  {
  "project": self.project.slug,
@@ -203,7 +203,7 @@ def test_simple_with_trends_p75(self, mock_get_trends):
  "trend_percentage": 0.88,
  }
  ]
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(
@@ -231,8 +231,8 @@ def test_simple_with_trends_p75(self, mock_get_trends):
  assert len(result_stats) > 0
  assert len(result_stats.get(f"{self.project.slug},foo", [])) > 0
 
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
- def test_simple_with_trends_p95(self, mock_get_trends):
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
+ def test_simple_with_trends_p95(self, mock_detect_breakpoints):
  mock_trends_result = [
  {
  "project": self.project.slug,
@@ -242,7 +242,7 @@ def test_simple_with_trends_p95(self, mock_get_trends):
  "trend_percentage": 0.88,
  }
  ]
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(
@@ -270,8 +270,8 @@ def test_simple_with_trends_p95(self, mock_get_trends):
  assert len(result_stats) > 0
  assert len(result_stats.get(f"{self.project.slug},foo", [])) > 0
 
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
- def test_simple_with_top_events(self, mock_get_trends):
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
+ def test_simple_with_top_events(self, mock_detect_breakpoints):
  # store second metric but with lower count
  self.store_performance_metric(
  name=TransactionMRI.DURATION.value,
@@ -300,7 +300,7 @@ def test_simple_with_top_events(self, mock_get_trends):
 
  assert response.status_code == 200, response.content
 
- trends_call_args_data = mock_get_trends.call_args[0][0]["data"]
+ trends_call_args_data = mock_detect_breakpoints.call_args[0][0]["data"]
  assert len(trends_call_args_data.get(f"{self.project.slug},foo")) > 0
  # checks that second transaction wasn't sent to the trends microservice
  assert len(trends_call_args_data.get(f"{self.project.slug},bar", [])) == 0
@@ -309,9 +309,9 @@ def test_simple_with_top_events(self, mock_get_trends):
  {"organizations:issue-platform": True, "organizations:performance-trends-issues": False}
  )
  @mock.patch("sentry.api.endpoints.organization_events_trends_v2.produce_occurrence_to_kafka")
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
  def test_skipped_issue_creation_no_feature_flag(
- self, mock_get_trends, mock_produce_occurrence_to_kafka
+ self, mock_detect_breakpoints, mock_produce_occurrence_to_kafka
  ):
  mock_trends_result = [
  {
@@ -321,7 +321,7 @@ def test_skipped_issue_creation_no_feature_flag(
  "trend_percentage": 2.0,
  }
  ]
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(
@@ -344,9 +344,9 @@ def test_skipped_issue_creation_no_feature_flag(
  {"organizations:issue-platform": True, "organizations:performance-trends-issues": True}
  )
  @mock.patch("sentry.api.endpoints.organization_events_trends_v2.produce_occurrence_to_kafka")
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
  def test_skipped_issue_creation_wrong_stats_period(
- self, mock_get_trends, mock_produce_occurrence_to_kafka
+ self, mock_detect_breakpoints, mock_produce_occurrence_to_kafka
  ):
  mock_trends_result = [
  {
@@ -356,7 +356,7 @@ def test_skipped_issue_creation_wrong_stats_period(
  "trend_percentage": 2.0,
  }
  ]
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(
@@ -379,9 +379,9 @@ def test_skipped_issue_creation_wrong_stats_period(
  {"organizations:issue-platform": True, "organizations:performance-trends-issues": True}
  )
  @mock.patch("sentry.api.endpoints.organization_events_trends_v2.produce_occurrence_to_kafka")
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
  def test_skipped_issue_creation_too_small_trend_percentage(
- self, mock_get_trends, mock_produce_occurrence_to_kafka
+ self, mock_detect_breakpoints, mock_produce_occurrence_to_kafka
  ):
  mock_trends_result = [
  {
@@ -392,7 +392,7 @@ def test_skipped_issue_creation_too_small_trend_percentage(
  "trend_percentage": 1.2,
  }
  ]
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(
@@ -415,9 +415,9 @@ def test_skipped_issue_creation_too_small_trend_percentage(
  {"organizations:issue-platform": True, "organizations:performance-trends-issues": True}
  )
  @mock.patch("sentry.api.endpoints.organization_events_trends_v2.produce_occurrence_to_kafka")
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
  def test_skipped_issue_creation_no_regression(
- self, mock_get_trends, mock_produce_occurrence_to_kafka
+ self, mock_detect_breakpoints, mock_produce_occurrence_to_kafka
  ):
  mock_trends_result = [
  {
@@ -427,7 +427,7 @@ def test_skipped_issue_creation_no_regression(
  "trend_percentage": 2.0,
  }
  ]
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(
@@ -450,9 +450,9 @@ def test_skipped_issue_creation_no_regression(
  {"organizations:issue-platform": True, "organizations:performance-trends-issues": True}
  )
  @mock.patch("sentry.api.endpoints.organization_events_trends_v2.produce_occurrence_to_kafka")
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
  def test_skipped_issue_creation_wrong_metric(
- self, mock_get_trends, mock_produce_occurrence_to_kafka
+ self, mock_detect_breakpoints, mock_produce_occurrence_to_kafka
  ):
  mock_trends_result = [
  {
@@ -462,7 +462,7 @@ def test_skipped_issue_creation_wrong_metric(
  "trend_percentage": 2.0,
  }
  ]
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(
@@ -485,8 +485,8 @@ def test_skipped_issue_creation_wrong_metric(
  {"organizations:issue-platform": True, "organizations:performance-trends-issues": True}
  )
  @mock.patch("sentry.api.endpoints.organization_events_trends_v2.produce_occurrence_to_kafka")
- @mock.patch("sentry.api.endpoints.organization_events_trends_v2.get_trends")
- def test_issue_creation_simple(self, mock_get_trends, mock_produce_occurrence_to_kafka):
+ @mock.patch("sentry.api.endpoints.organization_events_trends_v2.detect_breakpoints")
+ def test_issue_creation_simple(self, mock_detect_breakpoints, mock_produce_occurrence_to_kafka):
  mock_trends_result = [
  {
  "project": self.project.slug,
@@ -497,7 +497,7 @@ def test_issue_creation_simple(self, mock_get_trends, mock_produce_occurrence_to
  "aggregate_range_2": 28,
  }
  ]
- mock_get_trends.return_value = {"data": mock_trends_result}
+ mock_detect_breakpoints.return_value = {"data": mock_trends_result}
 
  with self.feature(self.features):
  response = self.client.get(