ref(statistical-detectors): Clean up statistical detectors calculation (#54694)

Clean up the calculations so that it's able to process things 1 at a
time instead of an arbitrary grouping.

Author: Zylphrex

src/sentry/statistical_detectors/detector.py

@@ -4,11 +4,8 @@
 from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
-from typing import Any, List, Mapping, MutableMapping, Optional, Tuple
+from typing import Any, Mapping, MutableMapping, Optional, Tuple
 
-from sentry_redis_tools.clients import RedisCluster, StrictRedis
-
-from sentry.models.project import Project
 from sentry.utils.math import ExponentialMovingAverage
 
 logger = logging.getLogger("sentry.tasks.statistical_detectors")
@@ -79,68 +76,25 @@ class TrendPayload:
     timestamp: datetime
 
 
-def run_trend_detection(
-    client: RedisCluster | StrictRedis,
-    project: Project,
-    start: datetime,
-    payloads: List[TrendPayload],
-) -> Tuple[List[TrendPayload], List[TrendPayload]]:
-    with client.pipeline() as pipeline:
-        for payload in payloads:
-            key = make_key(project.id, payload, VERSION)
-            pipeline.hgetall(key)
-        results = pipeline.execute()
-
-    old_states = [TrendState.from_dict(result) for result in results]
-    new_states, regressed, improved = compute_new_trend_states(project.id, old_states, payloads)
-
-    with client.pipeline() as pipeline:
-        for key, value in new_states:
-            pipeline.hmset(key, value.as_dict())
-            pipeline.expire(key, KEY_TTL)
-
-        pipeline.execute()
-
-    return regressed, improved
-
-
 def compute_new_trend_states(
-    project_id: int,
-    old_states: List[TrendState],
-    payloads: List[TrendPayload],
-) -> Tuple[List[Tuple[str, TrendState]], List[TrendPayload], List[TrendPayload]]:
-    new_states: List[Tuple[str, TrendState]] = []
-    regressed: List[TrendPayload] = []
-    improved: List[TrendPayload] = []
-
-    for payload, old_state in zip(payloads, old_states):
-        if old_state.timestamp is not None and old_state.timestamp > payload.timestamp:
-            # In the event that the timestamp is before the payload's timestamps,
-            # we do not want to process this payload.
-            #
-            # This should not happen other than in some error state.
-            logger.warning(
-                "Trend detection out of order. Processing %s, but last processed was %s",
-                payload.timestamp.isoformat(),
-                old_state.timestamp.isoformat(),
-            )
-            continue
-
-        key = make_key(project_id, payload, VERSION)
-        trend, value = detect_trend(old_state, payload)
-
-        if trend == TrendType.Regressed:
-            regressed.append(payload)
-        elif trend == TrendType.Improved:
-            improved.append(payload)
-
-        new_states.append((key, value))
-
-    return new_states, regressed, improved
-
-
-def make_key(project_id: int, payload: TrendPayload, version: int) -> str:
-    return f"statdtr:v:{version}:p:{project_id}:f:{payload.group}"
+    cur_state: TrendState,
+    payload: TrendPayload,
+) -> Optional[Tuple[TrendState, Tuple[TrendType, TrendPayload]]]:
+    if cur_state.timestamp is not None and cur_state.timestamp > payload.timestamp:
+        # In the event that the timestamp is before the payload's timestamps,
+        # we do not want to process this payload.
+        #
+        # This should not happen other than in some error state.
+        logger.warning(
+            "Trend detection out of order. Processing %s, but last processed was %s",
+            payload.timestamp.isoformat(),
+            cur_state.timestamp.isoformat(),
+        )
+        return None
+
+    trend, new_state = detect_trend(cur_state, payload)
+
+    return new_state, (trend, payload)
 
 
 def detect_trend(state: TrendState, payload: TrendPayload) -> Tuple[TrendType, TrendState]:

tests/sentry/statistical_detectors/test_detector.py

@@ -2,7 +2,12 @@
 
 import pytest
 
-from sentry.statistical_detectors.detector import TrendPayload, TrendState, compute_new_trend_states
+from sentry.statistical_detectors.detector import (
+    TrendPayload,
+    TrendState,
+    TrendType,
+    compute_new_trend_states,
+)
 
 
 @pytest.mark.parametrize(
@@ -101,7 +106,7 @@ def test_trend_state(data, expected):
     ],
 )
 def test_run_trend_detection(initial, values, regressed_indices, improved_indices):
-    states = [initial]
+    state: TrendState = initial
     all_regressed = []
     all_improved = []
 
@@ -113,10 +118,26 @@ def test_run_trend_detection(initial, values, regressed_indices, improved_indice
     ]
 
     for payload in payloads:
-        new_states, regressed, improved = compute_new_trend_states(1, states, [payload])
-        states = [state for _, state in new_states]
-        all_regressed.extend(regressed)
-        all_improved.extend(improved)
+        new_state = compute_new_trend_states(state, payload)
+        assert new_state is not None
+        state, (trend_type, result) = new_state
+        if trend_type == TrendType.Regressed:
+            all_regressed.append(result)
+        elif trend_type == TrendType.Improved:
+            all_improved.append(result)
 
     assert all_regressed == [payloads[i] for i in regressed_indices]
     assert all_improved == [payloads[i] for i in improved_indices]
+
+
+def test_run_trend_detection_bad_order():
+    now = datetime.now()
+
+    state = TrendState(None, 0, 0, 0)
+
+    new_state = compute_new_trend_states(state, TrendPayload(0, 2, 100, now))
+    assert new_state is not None
+    state, (trend_type, result) = new_state
+
+    new_state = compute_new_trend_states(state, TrendPayload(0, 1, 100, now - timedelta(hours=1)))
+    assert new_state is None