feat(alerts): add a regular job to detect anomalies

ee/tasks/test/subscriptions/test_subscriptions.py

@@ -15,7 +15,6 @@
 from posthog.models.exported_asset import ExportedAsset
 from posthog.models.insight import Insight
 from posthog.models.instance_setting import set_instance_setting
-from posthog.models.subscription import Subscription
 from posthog.test.base import APIBaseTest
 
 
@@ -24,7 +23,6 @@
 @patch("ee.tasks.subscriptions.generate_assets")
 @freeze_time("2022-02-02T08:55:00.000Z")
 class TestSubscriptionsTasks(APIBaseTest):
-    subscriptions: list[Subscription] = None  # type: ignore
     dashboard: Dashboard
     insight: Insight
     tiles: list[DashboardTile] = None  # type: ignore

frontend/src/lib/components/Alerts/views/EditAlert.tsx

@@ -70,10 +70,18 @@ export function EditAlert({ id, insightShortId, onCancel, onDelete }: EditAlertP
                         </LemonField>
                         <Group name={['anomaly_condition', 'absoluteThreshold']}>
                             <span className="flex gap-10">
-                                <LemonField name="lower" label="Lower threshold">
+                                <LemonField
+                                    name="lower"
+                                    label="Lower threshold"
+                                    help="Notify if the value is strictly below"
+                                >
                                     <LemonInput type="number" className="w-20" data-attr="alert-lower-threshold" />
                                 </LemonField>
-                                <LemonField name="upper" label="Upper threshold">
+                                <LemonField
+                                    name="upper"
+                                    label="Upper threshold"
+                                    help="Notify if the value is strictly above"
+                                >
                                     <LemonInput type="number" className="w-20" data-attr="alert-upper-threshold" />
                                 </LemonField>
                             </span>

frontend/src/queries/schema.json

@@ -1,6 +1,18 @@
 {
     "$schema": "http://json-schema.org/draft-07/schema#",
     "definitions": {
+        "AbsoluteThreshold": {
+            "additionalProperties": false,
+            "properties": {
+                "lower": {
+                    "type": ["number", "null"]
+                },
+                "upper": {
+                    "type": ["number", "null"]
+                }
+            },
+            "type": "object"
+        },
         "ActionsNode": {
             "additionalProperties": false,
             "properties": {
@@ -179,6 +191,16 @@
             "enum": ["numeric", "duration", "duration_ms", "percentage", "percentage_scaled"],
             "type": "string"
         },
+        "AnomalyCondition": {
+            "additionalProperties": false,
+            "properties": {
+                "absoluteThreshold": {
+                    "$ref": "#/definitions/AbsoluteThreshold"
+                }
+            },
+            "required": ["absoluteThreshold"],
+            "type": "object"
+        },
         "AnyDataNode": {
             "anyOf": [
                 {

frontend/src/queries/schema.ts

@@ -1584,3 +1584,12 @@ export interface DashboardFilter {
     date_to?: string | null
     properties?: AnyPropertyFilter[] | null
 }
+
+export interface AbsoluteThreshold {
+    lower?: number | null
+    upper?: number | null
+}
+
+export interface AnomalyCondition {
+    absoluteThreshold: AbsoluteThreshold
+}

frontend/src/scenes/insights/InsightNav/insightNavLogic.tsx

@@ -274,7 +274,7 @@ export const insightNavLogic = kea<insightNavLogicType>([
         },
     })),
     urlToAction(({ actions }) => ({
-        '/insights/:shortId(/:mode)(/:subscriptionId)': (
+        '/insights/:shortId(/:mode)(/:itemId)': (
             _, // url params
             { dashboard, ...searchParams }, // search params
             { filters: _filters } // hash params

frontend/src/scenes/insights/InsightPageHeader.tsx

@@ -33,7 +33,7 @@ import { ExporterFormat, InsightLogicProps, ItemMode, NotebookNodeType } from '~
 
 export function InsightPageHeader({ insightLogicProps }: { insightLogicProps: InsightLogicProps }): JSX.Element {
     // insightSceneLogic
-    const { insightMode, subscriptionId } = useValues(insightSceneLogic)
+    const { insightMode, itemId } = useValues(insightSceneLogic)
     const { setInsightMode } = useActions(insightSceneLogic)
 
     // insightLogic
@@ -71,7 +71,7 @@ export function InsightPageHeader({ insightLogicProps }: { insightLogicProps: In
                         isOpen={insightMode === ItemMode.Subscriptions}
                         closeModal={() => push(urls.insightView(insight.short_id))}
                         insightShortId={insight.short_id}
-                        subscriptionId={subscriptionId}
+                        subscriptionId={itemId}
                     />
                     <SharingModal
                         title="Insight sharing"
@@ -91,7 +91,7 @@ export function InsightPageHeader({ insightLogicProps }: { insightLogicProps: In
                         isOpen={insightMode === ItemMode.Alerts}
                         closeModal={() => push(urls.insightView(insight.short_id))}
                         insightShortId={insight.short_id}
-                        alertId={subscriptionId}
+                        alertId={itemId}
                     />
                     <NewDashboardModal />
                 </>

frontend/src/scenes/insights/insightSceneLogic.tsx

@@ -34,10 +34,10 @@ export const insightSceneLogic = kea<insightSceneLogicType>([
     actions({
         setInsightId: (insightId: InsightShortId) => ({ insightId }),
         setInsightMode: (insightMode: ItemMode, source: InsightEventSource | null) => ({ insightMode, source }),
-        setSceneState: (insightId: InsightShortId, insightMode: ItemMode, subscriptionId: string | undefined) => ({
+        setSceneState: (insightId: InsightShortId, insightMode: ItemMode, itemId: string | undefined) => ({
             insightId,
             insightMode,
-            subscriptionId,
+            itemId,
         }),
         setInsightLogicRef: (logic: BuiltLogic<insightLogicType> | null, unmount: null | (() => void)) => ({
             logic,
@@ -61,15 +61,11 @@ export const insightSceneLogic = kea<insightSceneLogicType>([
                 setSceneState: (_, { insightMode }) => insightMode,
             },
         ],
-        subscriptionId: [
+        itemId: [
             null as null | number | 'new',
             {
-                setSceneState: (_, { subscriptionId }) =>
-                    subscriptionId !== undefined
-                        ? subscriptionId === 'new'
-                            ? 'new'
-                            : parseInt(subscriptionId, 10)
-                        : null,
+                setSceneState: (_, { itemId }) =>
+                    itemId !== undefined ? (itemId === 'new' ? 'new' : parseInt(itemId, 10)) : null,
             },
         ],
         insightLogicRef: [
@@ -210,8 +206,8 @@ export const insightSceneLogic = kea<insightSceneLogicType>([
                 source: JSON.parse(q),
             } as DataVisualizationNode)
         },
-        '/insights/:shortId(/:mode)(/:subscriptionId)': (
-            { shortId, mode, subscriptionId }, // url params
+        '/insights/:shortId(/:mode)(/:itemId)': (
+            { shortId, mode, itemId }, // url params
             { dashboard, ...searchParams }, // search params
             { filters: _filters, q }, // hash params
             { method, initial }, // "location changed" event payload
@@ -245,12 +241,8 @@ export const insightSceneLogic = kea<insightSceneLogicType>([
                 return
             }
 
-            if (
-                insightId !== values.insightId ||
-                insightMode !== values.insightMode ||
-                subscriptionId !== values.subscriptionId
-            ) {
-                actions.setSceneState(insightId, insightMode, subscriptionId)
+            if (insightId !== values.insightId || insightMode !== values.insightMode || itemId !== values.itemId) {
+                actions.setSceneState(insightId, insightMode, itemId)
             }
 
             // capture any filters from the URL, either #filters={} or ?insight=X&bla=foo&bar=baz

frontend/src/scenes/scenes.ts

@@ -510,8 +510,8 @@ export const routes: Record<string, Scene> = {
     [urls.insightEdit(':shortId' as InsightShortId)]: Scene.Insight,
     [urls.insightView(':shortId' as InsightShortId)]: Scene.Insight,
     [urls.insightSubcriptions(':shortId' as InsightShortId)]: Scene.Insight,
-    [urls.insightSubcription(':shortId' as InsightShortId, ':subscriptionId')]: Scene.Insight,
-    [urls.alert(':shortId' as InsightShortId, ':subscriptionId')]: Scene.Insight,
+    [urls.insightSubcription(':shortId' as InsightShortId, ':itemId')]: Scene.Insight,
+    [urls.alert(':shortId' as InsightShortId, ':itemId')]: Scene.Insight,
     [urls.alerts(':shortId' as InsightShortId)]: Scene.Insight,
     [urls.insightSharing(':shortId' as InsightShortId)]: Scene.Insight,
     [urls.savedInsights()]: Scene.SavedInsights,

frontend/src/types.ts

@@ -4320,6 +4320,7 @@ export type HogFunctionInvocationGlobals = {
     >
 }
 
+// TODO: move to schema.ts
 export interface AnomalyCondition {
     absoluteThreshold: {
         lower?: number

posthog/caching/calculate_results.py

@@ -123,7 +123,7 @@ def get_cache_type(cacheable: Optional[FilterType] | Optional[dict]) -> CacheTyp
 
 
 def calculate_for_query_based_insight(
-    insight: Insight, *, dashboard: Optional[Dashboard] = None, execution_mode: ExecutionMode, user: User
+    insight: Insight, *, dashboard: Optional[Dashboard] = None, execution_mode: ExecutionMode, user: Optional[User]
 ) -> "InsightResult":
     from posthog.caching.fetch_from_cache import InsightResult, NothingInCacheResult
     from posthog.caching.insight_cache import update_cached_state

posthog/models/alert.py

@@ -3,7 +3,7 @@
 
 class Alert(models.Model):
     team: models.ForeignKey = models.ForeignKey("Team", on_delete=models.CASCADE)
-    insight = models.ForeignKey("posthog.Insight", on_delete=models.CASCADE)
+    insight: models.ForeignKey = models.ForeignKey("posthog.Insight", on_delete=models.CASCADE)
 
     name: models.CharField = models.CharField(max_length=100)
     target_value: models.TextField = models.TextField()

posthog/schema.py

@@ -12,6 +12,14 @@ class SchemaRoot(RootModel[Any]):
     root: Any
 
 
+class AbsoluteThreshold(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    lower: Optional[float] = None
+    upper: Optional[float] = None
+
+
 class MathGroupTypeIndex(float, Enum):
     NUMBER_0 = 0
     NUMBER_1 = 1
@@ -28,6 +36,13 @@ class AggregationAxisFormat(StrEnum):
     PERCENTAGE_SCALED = "percentage_scaled"
 
 
+class AnomalyCondition(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    absoluteThreshold: AbsoluteThreshold
+
+
 class Kind(StrEnum):
     METHOD = "Method"
     FUNCTION = "Function"

posthog/tasks/alerts/checks.py

@@ -0,0 +1,111 @@
+from celery import shared_task
+from celery.canvas import group, chain
+from django.utils import timezone
+import math
+import structlog
+from typing import Any
+
+from posthog.api.services.query import ExecutionMode
+from posthog.caching.calculate_results import calculate_for_query_based_insight
+from posthog.email import EmailMessage
+from posthog.hogql_queries.legacy_compatibility.flagged_conversion_manager import (
+    conversion_to_query_based,
+)
+from posthog.models import Alert
+from posthog.schema import AnomalyCondition
+
+logger = structlog.get_logger(__name__)
+
+
+def check_all_alerts() -> None:
+    alert_ids = list(Alert.objects.all().values_list("id", flat=True))
+
+    group_count = 10
+    # All groups but the last one will have a group_size size.
+    # The last group will have at most group_size size.
+    group_size = int(math.ceil(len(alert_ids) / group_count))
+
+    groups = []
+    for i in range(0, len(alert_ids), group_size):
+        alert_id_group = alert_ids[i : i + group_size]
+        chained_calls = chain([check_alert_task.s(alert_id=alert_id) for alert_id in alert_id_group])
+        groups.append(chained_calls)
+
+    group(groups).apply_async()
+
+
+def check_alert(alert_id: int) -> None:
+    alert = Alert.objects.get(pk=alert_id)
+    insight = alert.insight
+
+    with conversion_to_query_based(insight):
+        calculation_result = calculate_for_query_based_insight(
+            insight,
+            execution_mode=ExecutionMode.RECENT_CACHE_CALCULATE_BLOCKING_IF_STALE,
+            user=None,
+        )
+
+    if not calculation_result.result:
+        raise RuntimeError(f"No results for alert {alert.id}")
+
+    anomaly_condition = AnomalyCondition.model_validate(alert.anomaly_condition)
+    thresholds = anomaly_condition.absoluteThreshold
+
+    result = calculation_result.result[0]
+    aggregated_value = result["aggregated_value"]
+    anomalies_descriptions = []
+
+    if thresholds.lower is not None and aggregated_value < thresholds.lower:
+        anomalies_descriptions += [
+            f"The trend value ({aggregated_value}) is below the lower threshold ({thresholds.lower})"
+        ]
+    if thresholds.upper is not None and aggregated_value > thresholds.upper:
+        anomalies_descriptions += [
+            f"The trend value ({aggregated_value}) is above the upper threshold ({thresholds.upper})"
+        ]
+
+    if not anomalies_descriptions:
+        logger.info("No threshold met", alert_id=alert.id)
+        return
+
+    send_notifications(alert, anomalies_descriptions)
+
+
+@shared_task(ignore_result=True)
+def check_all_alerts_task() -> None:
+    check_all_alerts()
+
+
+# Note, check_alert_task is used in Celery chains. Celery chains pass the previous
+# function call result to the next function as an argument, hence args and kwargs.
+@shared_task(ignore_result=True)
+def check_alert_task(*args: Any, **kwargs: Any) -> None:
+    check_alert(**kwargs)
+
+
+# TODO: make it a task
+def send_notifications(alert: Alert, anomalies_descriptions: list[str]) -> None:
+    subject = f"PostHog alert {alert.name} has anomalies"
+    campaign_key = f"alert-anomaly-notification-{alert.id}-{timezone.now().isoformat()}"
+    insight_url = f"/project/{alert.team.pk}/insights/{alert.insight.short_id}"
+    alert_url = f"{insight_url}/alerts/{alert.id}"
+    message = EmailMessage(
+        campaign_key=campaign_key,
+        subject=subject,
+        template_name="alert_anomaly",
+        template_context={
+            "anomalies_descriptions": anomalies_descriptions,
+            "insight_url": insight_url,
+            "insight_name": alert.insight.name,
+            "alert_url": alert_url,
+            "alert_name": alert.name,
+        },
+    )
+    targets = list(filter(len, alert.target_value.split(",")))
+    if not targets:
+        raise RuntimeError(f"no targets configured for the alert {alert.id}")
+    for target in targets:
+        message.add_recipient(email=target)
+
+    logger.info(f"Send notifications about {len(anomalies_descriptions)} anomalies", alert_id=alert.id)
+    message.send()