Event detector slack integration

dags/pipeline.py

@@ -63,15 +63,24 @@ def run_make_analysis(
 
 
 def run_make_event_detector(
-    clickhouse_url: str, probe_cc: List[str], timestamp: str = "", ts: str = ""
+    clickhouse_url: str,
+    probe_cc: List[str],
+    timestamp: str = "",
+    ts: str = "",
+    slack_webhook: str | None = None,
+    explorer_base_url: str = "https://explorer.ooni.org/",
 ):
     from oonipipeline.tasks.detector import MakeDetectorParams, make_detector
 
     if timestamp == "":
         timestamp = ts[:13]
 
     params = MakeDetectorParams(
-        clickhouse_url=clickhouse_url, probe_cc=probe_cc, timestamp=timestamp
+        clickhouse_url=clickhouse_url,
+        probe_cc=probe_cc,
+        timestamp=timestamp,
+        slack_webhook=slack_webhook,
+        explorer_base_url=explorer_base_url,
     )
 
     make_detector(params)
@@ -229,6 +238,10 @@ def run_make_time_inconsistencies_analysis(
         op_kwargs={
             "probe_cc": dag_full.params["probe_cc"],
             "clickhouse_url": Variable.get("clickhouse_url", default_var=""),
+            "slack_webhook": Variable.get("slack_webhook", default_var=None),
+            "explorer_base_url": Variable.get(
+                "explorer_base_url", default_var="https://explorer.ooni.org/"
+            ),
             "ts": "{{ts}}",
         },
         requirements=REQUIREMENTS,

oonipipeline/src/oonipipeline/analysis/detector.py

@@ -3,6 +3,8 @@
 from enum import Enum
 from itertools import groupby as itertools_groupby
 from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple
+from urllib.parse import urlencode
+import requests
 
 from clickhouse_driver import Client as ClickhouseClient
 
@@ -507,6 +509,8 @@ def run_detector(
     gap_halflife: float = 48.0,
     warmup: bool = False,
     trace: bool = False,
+    slack_webhook: str | None = None,
+    explorer_base_url: str = "https://explorer.ooni.org/",
 ) -> Tuple[List[Changepoint], List[LastCusum], List[CusumStep]]:
     db = ClickhouseClient.from_url(clickhouse_url)
     domains = get_domain_list(db)
@@ -531,6 +535,9 @@ def run_detector(
     )
     update_tables(db, updated_cusums, changepoints)
 
+    if slack_webhook is not None:
+        notify_slack(changepoints, slack_webhook, explorer_base_url)
+
     return changepoints, updated_cusums, steps
 
 
@@ -649,3 +656,95 @@ def make_label(df, field, label, color):
         .interactive()
     )
     chart.show()
+
+
+def notify_slack(
+    changepoints: list[Changepoint],
+    slack_webhook: str,
+    explorer_base_url: str = "https://explorer.ooni.org/",
+):
+    """
+    Sends a message to slack with a list of all changepoints that were detected
+    in the last run
+    """
+
+    if len(changepoints) == 0:
+        return
+
+    message = (
+        "*NEW EVENTS DETECTED*\n\nWe just detected the following blocking events:\n"
+    )
+
+    def dir_to_str(dir: int) -> str:
+        to_str = {
+            -1: "is unblocked :arrow_down: :large_green_circle:",
+            1: "is blocked :arrow_up: :red_circle:",
+        }
+        return to_str.get(dir, f"Unknown direction change value: {dir} :thinking_face:")
+
+    messages = []
+    for i, cp in enumerate(changepoints):
+        explorer = get_explorer_url(cp, explorer_base_url)
+        # Alerts panel not yet deployed to prod, we use the test one for now
+        alerts = get_alert_page_url(cp, "https://explorer.test.ooni.org/")
+        message += (
+            f"• :flag-{cp['probe_cc'].lower()}: [{cp['probe_cc']}/AS{cp['probe_asn']}] "
+            f"*{cp['domain']}* {dir_to_str(cp['change_dir'])} - `{cp['block_type']}` "
+            f"| <{explorer}|explorer> | <{alerts}|alerts>\n"
+        )
+
+        # Send messages in 10 entries batches to avoid max message size limit
+        if (i + 1) % 10 == 0:
+            messages.append(message)
+            message = ""
+
+    if message != "":
+        messages.append(message)
+
+    # Send messages to slack
+    for msg in messages:
+        send_to_slack(slack_webhook, msg)
+
+
+def send_to_slack(webhook: str, message: str):
+    requests.post(webhook, json={"text": message}).raise_for_status()
+
+
+def get_explorer_url(
+    changepoint: Changepoint, base_url: str = "https://explorer.ooni.org/"
+) -> str:
+    start_time = changepoint["ts"] - timedelta(days=13)
+    end_time = changepoint["ts"] + timedelta(days=2)
+
+    def to_s(dt: datetime):
+        return datetime.strftime(dt, "%Y-%m-%d")
+
+    params = {
+        "domain": changepoint["domain"],
+        "probe_cc": changepoint["probe_cc"],
+        "probe_asn": changepoint["probe_asn"],
+        "since": to_s(start_time),
+        "until": to_s(end_time),
+        "axis_x": "measurement_start_day",
+    }
+    url = f"{base_url}/chart/mat?{urlencode(params)}"
+    return url
+
+def get_alert_page_url(
+    changepoint: Changepoint, base_url: str = "https://explorer.ooni.org/"
+) -> str:
+    start_time = changepoint["ts"] - timedelta(days=13)
+    end_time = changepoint["ts"] + timedelta(days=2)
+
+    def to_s(dt: datetime):
+        return datetime.strftime(dt, "%Y-%m-%d")
+
+    params = {
+        "domain": changepoint["domain"],
+        "probe_cc": changepoint["probe_cc"],
+        "probe_asn": changepoint["probe_asn"],
+        "since": to_s(start_time),
+        "until": to_s(end_time),
+    }
+    url = f"{base_url}/chart/alerts?{urlencode(params)}"
+    return url

oonipipeline/src/oonipipeline/tasks/detector.py

@@ -10,6 +10,8 @@ class MakeDetectorParams:
     clickhouse_url: str
     probe_cc: List[str]
     timestamp: str
+    slack_webhook: str | None = None
+    explorer_base_url: str = "https://explorer.ooni.org/"
 
 
 def make_detector(params: MakeDetectorParams):
@@ -29,4 +31,6 @@ def make_detector(params: MakeDetectorParams):
         start_time=start_hour,
         end_time=end_hour,
         probe_cc=params.probe_cc,
+        slack_webhook=params.slack_webhook,
+        explorer_base_url=params.explorer_base_url,
     )

oonipipeline/tests/test_e2e.py

@@ -123,7 +123,7 @@ def test_event_detector(datadir, db):
         )
         make_detector(
             MakeDetectorParams(
-                clickhouse_url=db.clickhouse_url, probe_cc=["TG"], timestamp=timestamp
+                clickhouse_url=db.clickhouse_url, probe_cc=["TG"], timestamp=timestamp, slack_webhook=None
             )
         )
     res = db.execute("SELECT COUNT() FROM event_detector_changepoints")