diff --git a/dags/pipeline.py b/dags/pipeline.py index 7fba8e07..17d78f25 100644 --- a/dags/pipeline.py +++ b/dags/pipeline.py @@ -63,7 +63,12 @@ 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 @@ -71,7 +76,11 @@ def run_make_event_detector( 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, diff --git a/oonipipeline/src/oonipipeline/analysis/detector.py b/oonipipeline/src/oonipipeline/analysis/detector.py index 05c885ab..3dbec55e 100644 --- a/oonipipeline/src/oonipipeline/analysis/detector.py +++ b/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 diff --git a/oonipipeline/src/oonipipeline/tasks/detector.py b/oonipipeline/src/oonipipeline/tasks/detector.py index 1e0a0e90..d5d5d71f 100644 --- a/oonipipeline/src/oonipipeline/tasks/detector.py +++ b/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, ) diff --git a/oonipipeline/tests/test_e2e.py b/oonipipeline/tests/test_e2e.py index 6f26780b..8881b0d3 100644 --- a/oonipipeline/tests/test_e2e.py +++ b/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")