diff --git a/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/__init__.py b/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/constants.py b/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/constants.py deleted file mode 100644 index fd0aeb17..00000000 --- a/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/constants.py +++ /dev/null @@ -1,90 +0,0 @@ -# -*- coding: utf-8 -*- -from enum import Enum - - -class Querys(Enum): - FLOWS_FAILED_LAST_WEEK = """ - query($since: timestamptz!) { - flow( - where: { - schedule: { _is_null: false } - is_schedule_active: {_eq: true} - archived: {_eq: false} - flow_runs: { - state: { _eq: "Failed" } - start_time: { _gte: $since } - } - } - ) { - id - created - name - } - } - """ - - ACTIVE_FLOWS_BY_NAMES = """ - query($names: [String!]!) { - flow( - where: { - name: { _in: $names } - is_schedule_active: { _eq: true } - archived: { _eq: false } - } - ) { - id - name - } - } - """ - - LAST_COMPLETED_RUNS_TASKS = """ - query LastTwoCompletedRunsWithTasks($flow_id: uuid!) { - flow_run( - where: { - flow_id: { _eq: $flow_id } - state: { _in: ["Success", "Failed"] } - start_time: { _is_null: false } - } - order_by: { start_time: desc } - limit: 2 - ) { - id - name - start_time - state - task_runs( - where: { - state: { _in: ["Failed"] } - } - order_by: { start_time: desc } - limit: 1) { - id - state - end_time - state_message - task { - id - name - } - } - } - } - """ - - -class Constants(Enum): - TASKS_NAME_DISABLE = ("run_dbt",) - FLOW_SUCCESS_STATE = "Success" - FLOW_FAILED_STATE = "Failed" - - PREFECT_URL = "https://prefect.basedosdados.org/" - PREFECT_URL_FLOW = PREFECT_URL + "flow/" - PREFECT_URL_API = PREFECT_URL + "api" - - DISCORD_ROLE_DADOS = "865034571469160458" - TEXT_FLOW_FORMAT = "- {run_name} | Last failure `{task_name}` | {link}" - - STATE_MESSAGE_IGNORE = ( - "No heartbeat detected from the remote task; marking the run as failed.", - ) diff --git a/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/datetime_utils.py b/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/datetime_utils.py deleted file mode 100644 index 3504c198..00000000 --- a/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/datetime_utils.py +++ /dev/null @@ -1,25 +0,0 @@ -# -*- coding: utf-8 -*- -from datetime import datetime, timedelta - - -def parse_datetime(value: str) -> datetime: - """Parse an ISO 8601 string into a datetime object. - - Args: - value: ISO 8601 formatted datetime string. - - Returns: - Parsed datetime object. - """ - return datetime.fromisoformat(value) - - -def one_week_ago() -> str: - """Return a timestamp string representing exactly one week ago. - - Returns: - UTC timestamp formatted as ``%Y-%m-%dT%H:%M:%SZ``. - """ - from django.utils import timezone - - return (timezone.now() - timedelta(days=7)).strftime("%Y-%m-%dT%H:%M:%SZ") diff --git a/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/models.py b/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/models.py deleted file mode 100644 index e8618119..00000000 --- a/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/models.py +++ /dev/null @@ -1,142 +0,0 @@ -# -*- coding: utf-8 -*- -from datetime import datetime -from typing import TYPE_CHECKING, Optional - -from .constants import Constants -from .datetime_utils import parse_datetime - -if TYPE_CHECKING: - from .service import FlowService - - -class Task: - """Represents a Prefect task.""" - - def __init__(self, id: str, name: str) -> None: - """Args: - id: Prefect task UUID. - name: Task name. - """ - self.id = id - self.name = name - - -class TaskRun: - """Represents a single run of a Prefect task.""" - - def __init__( - self, - id: str, - state: str, - end_time: Optional[str], - state_message: str, - task: dict, - ) -> None: - """Args: - id: Prefect task run UUID. - state: Run state (e.g. ``"Failed"``). - end_time: ISO 8601 end timestamp, or ``None`` if not finished. - state_message: Human-readable state message from Prefect. - task: Raw task dict with ``id`` and ``name`` keys. - """ - self.id = id - self.state = state - self.end_time = end_time - self.state_message = state_message - self.task = Task(**task) - - -class FlowRun: - """Represents a single run of a Prefect flow.""" - - def __init__( - self, - id: str, - name: str, - start_time: str, - state: str, - task_runs: list, - ) -> None: - """Args: - id: Prefect flow run UUID. - name: Flow run name. - start_time: ISO 8601 start timestamp. - state: Run state (e.g. ``"Success"`` or ``"Failed"``). - task_runs: List of raw task run dicts. Only the first entry is used. - """ - self.id = id - self.name = name - self.start_time = parse_datetime(start_time) - self.state = state - self.task_runs = TaskRun(**task_runs[0]) if task_runs else None - - -class FlowDisable: - """Represents a Prefect flow candidate for disabling. - - On instantiation, fetches the last two completed runs from the Prefect API - to support validation. - """ - - def __init__( - self, - id: str, - created: str | datetime, - service: "FlowService", - name: str = "", - ) -> None: - """Args: - id: Prefect flow UUID. - created: The earliest timestamp from which failures are considered - relevant — either an ISO 8601 string or a datetime. For new - flows this is Prefect's ``created`` field; for flows reactivated - by an admin, ``reactivated_at`` is passed instead. - service: ``FlowService`` instance used to fetch run data. - name: Flow name. Defaults to empty string. - """ - self.id = id - self.name = name - self.valid_since = parse_datetime(created) if isinstance(created, str) else created - self.service = service - self.runs = self.get_runs() - - def get_runs(self) -> list[FlowRun]: - """Fetch the last two completed runs for this flow from the Prefect API. - - Returns: - List of up to two ``FlowRun`` objects ordered by start time descending. - """ - response = self.service.last_completed_runs_tasks(self.id) - return [FlowRun(**run) for run in response["flow_run"]] - - def validate(self) -> bool: - """Determine whether this flow should be disabled. - - A flow is considered unhealthy — and should be disabled — when either - of the following conditions holds after ``self.valid_since``: - - - The last run failed on a ``run_dbt`` task with a non-ignorable error. - - The last two runs both failed. - - Returns: - ``True`` if the flow should be disabled, ``False`` otherwise. - """ - last_run = self.runs[0] - next_last = self.runs[1] if len(self.runs) == 2 else None - - failed = Constants.FLOW_FAILED_STATE.value - dbt_failed_after_created = ( - last_run.state == failed - and last_run.task_runs.task.name in Constants.TASKS_NAME_DISABLE.value - and last_run.start_time >= self.valid_since - and last_run.task_runs.state_message not in Constants.STATE_MESSAGE_IGNORE.value - ) - - consecutive_failed_after_created = ( - next_last - and last_run.state == failed - and next_last.state == failed - and max(last_run.start_time, next_last.start_time) >= self.valid_since - ) - - return bool(dbt_failed_after_created or consecutive_failed_after_created) diff --git a/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/service.py b/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/service.py deleted file mode 100644 index b818022d..00000000 --- a/backend/apps/admin_data_tools/management/commands/_disable_unhealthy_flow_schedules/service.py +++ /dev/null @@ -1,350 +0,0 @@ -# -*- coding: utf-8 -*- -import os -from typing import Dict - -from django.db.models import Q -from gql import Client, gql -from gql.transport.requests import RequestsHTTPTransport -from loguru import logger - -from backend.apps.admin_data_tools.models import DisabledFlowSchedule -from backend.custom.client import send_discord_message - -from .constants import Constants, Querys -from .datetime_utils import one_week_ago -from .models import FlowDisable - -logger = logger.bind(module="admin_data_tools") - - -class MakeClient: - """Builds an authenticated GraphQL client for the Prefect API.""" - - def __init__(self) -> None: - self.graphql_url = Constants.PREFECT_URL_API.value - self.query = self.make_client({"Authorization": f"Bearer {os.getenv('API_KEY_PREFECT')}"}) - - def make_client(self, headers: Dict[str, str] = None) -> Client: - """Instantiate a GQL ``Client`` with the given headers. - - Args: - headers: HTTP headers to attach to every request. - - Returns: - Configured ``gql.Client`` instance. - """ - transport = RequestsHTTPTransport(url=self.graphql_url, headers=headers, use_json=True) - return Client(transport=transport, fetch_schema_from_transport=False) - - -class FlowService: - """Orchestrates detection and disabling of unhealthy Prefect flow schedules.""" - - def __init__(self) -> None: - self.client = MakeClient() - - # ------------------------------------------------------------------ - # Prefect API helpers - # ------------------------------------------------------------------ - - def flows_failed_last_week(self) -> list[dict]: - """Query Prefect for active flows that had at least one failed run in the past week. - - Returns: - List of dicts with ``id``, ``created``, and ``name`` keys. - """ - since = one_week_ago() - response = self.client.query.execute( - gql(Querys.FLOWS_FAILED_LAST_WEEK.value), variable_values={"since": since} - ) - return [ - {"id": fail["id"], "created": fail["created"], "name": fail["name"]} - for fail in response["flow"] - ] - - def active_flows_by_names(self, names: list[str]) -> list[dict]: - """Query Prefect for flows that are currently active, filtered by name. - - Args: - names: List of flow names to look up. - - Returns: - List of dicts with ``id`` and ``name`` keys. - """ - response = self.client.query.execute( - gql(Querys.ACTIVE_FLOWS_BY_NAMES.value), variable_values={"names": names} - ) - return response["flow"] - - def last_completed_runs_tasks(self, flow_id: str) -> dict: - """Fetch the last two completed (Success or Failed) runs for a flow. - - Args: - flow_id: Prefect flow UUID. - - Returns: - Raw GraphQL response dict containing a ``flow_run`` list. - """ - return self.client.query.execute( - gql(Querys.LAST_COMPLETED_RUNS_TASKS.value), variable_values={"flow_id": flow_id} - ) - - def set_flow_schedule(self, flow_id: str, active: bool) -> dict: - """Toggle a flow's schedule on or off in Prefect. - - Args: - flow_id: Prefect flow UUID. - active: ``True`` to activate the schedule, ``False`` to deactivate. - - Returns: - Raw GraphQL response dict. - """ - mutation_name = "set_schedule_active" if active else "set_schedule_inactive" - query = f""" - mutation SetFlowSchedule($flow_id: UUID!) {{ - {mutation_name}( - input: {{ - flow_id: $flow_id - }} - ) {{ - success - }} - }} - """ - return self.client.query.execute(gql(query), variable_values={"flow_id": flow_id}) - - def disable_flow_schedule(self, flow_id: str) -> None: - """Disable a flow schedule, calling the mutation twice as a Prefect workaround. - - Note: - Prefect 0.15 has a bug where a single ``set_schedule_inactive`` call - is not always effective. Two consecutive calls are required. - - Args: - flow_id: Prefect flow UUID to disable. - """ - for _ in range(2): - self.set_flow_schedule(flow_id=flow_id, active=False) - - # ------------------------------------------------------------------ - # Main entry point - # ------------------------------------------------------------------ - - def disable_unhealthy_flow_schedules(self) -> None: - """Run both phases of the unhealthy-flow detection and disabling pipeline. - - Phase 1 — enforce known-disabled flows: re-disables flows that Prefect - reactivated (e.g. after re-registration) and detects new failures in - flows that an admin previously reactivated. - - Phase 2 — detect new unhealthy flows: queries Prefect for flows that - failed in the past week, validates each one, disables the unhealthy - ones, and sends a Discord notification. - """ - self._enforce_disabled_flows() - self._detect_and_disable_new_flows() - - # ------------------------------------------------------------------ - # Phase 1 - # ------------------------------------------------------------------ - - def _enforce_disabled_flows(self) -> None: - """Re-disable tracked flows that Prefect reactivated. - - Also checks reactivated flows for new failures since reactivation. - """ - all_tracked = list( - DisabledFlowSchedule.objects.filter( - Q(is_schedule_active=False) | Q(reactivated_at__isnull=False) - ) - ) - if not all_tracked: - return - - tracked_by_name = {record.flow_name: record for record in all_tracked} - active_in_prefect = self.active_flows_by_names(list(tracked_by_name.keys())) - active_by_name = {flow["name"]: flow["id"] for flow in active_in_prefect} - - for flow_name, record in tracked_by_name.items(): - if flow_name not in active_by_name: - continue - self._process_active_tracked_flow(record, active_by_name[flow_name]) - - def _process_active_tracked_flow( - self, record: DisabledFlowSchedule, current_flow_id: str - ) -> None: - """Sync the stored flow ID and apply the appropriate enforcement action. - - Args: - record: The ``DisabledFlowSchedule`` database record. - current_flow_id: The flow's current UUID in Prefect. - """ - self._sync_flow_id(record, current_flow_id) - - if not record.is_schedule_active: - self._re_disable_flow(record, current_flow_id) - elif record.reactivated_at: - self._check_post_reactivation(record, current_flow_id) - - def _sync_flow_id(self, record: DisabledFlowSchedule, current_flow_id: str) -> None: - """Update the stored flow ID if Prefect re-registered the flow with a new UUID. - - Args: - record: The ``DisabledFlowSchedule`` database record. - current_flow_id: The flow's current UUID in Prefect. - """ - if record.flow_id == current_flow_id: - return - record.flow_id = current_flow_id - record.save(update_fields=["flow_id"]) - - def _re_disable_flow(self, record: DisabledFlowSchedule, current_flow_id: str) -> None: - """Disable a flow that should remain inactive but was reactivated by Prefect. - - Args: - record: The ``DisabledFlowSchedule`` database record. - current_flow_id: The flow's current UUID in Prefect. - """ - self.disable_flow_schedule(flow_id=current_flow_id) - logger.info(f"Re-disabled flow {record.flow_name} ({current_flow_id})") - - def _check_post_reactivation(self, record: DisabledFlowSchedule, current_flow_id: str) -> None: - """Disable a flow if it broke again after being reactivated by an admin. - - Uses ``reactivated_at`` as the baseline timestamp instead of Prefect's - ``created`` field, which is unreliable after re-registration. - - Args: - record: The ``DisabledFlowSchedule`` database record. - current_flow_id: The flow's current UUID in Prefect. - """ - flow_disable = FlowDisable( - id=current_flow_id, - created=record.reactivated_at.isoformat(), - service=self, - ) - if not flow_disable.validate(): - return - self.disable_flow_schedule(flow_id=current_flow_id) - record.is_schedule_active = False - record.reactivated_at = None - record.save(update_fields=["is_schedule_active", "reactivated_at"]) - logger.info(f"Re-disabled flow {record.flow_name} after reactivation failure") - - # ------------------------------------------------------------------ - # Phase 2 - # ------------------------------------------------------------------ - - def _detect_and_disable_new_flows(self) -> None: - """Detect untracked unhealthy flows, disable them, and send a notification.""" - flows = self._get_new_untracked_flows() - flows_to_disable = [flow for flow in flows if flow.validate()] - - if not flows_to_disable: - return - - for flow in flows_to_disable: - self._disable_and_register(flow) - - self._send_disable_notification(flows, flows_to_disable) - - def _get_new_untracked_flows(self) -> list[FlowDisable]: - """Return flows that failed last week but are not yet tracked in the database. - - Returns: - List of ``FlowDisable`` instances for untracked failing flows. - """ - flows_data = self.flows_failed_last_week() - tracked_names = set(DisabledFlowSchedule.objects.values_list("flow_name", flat=True)) - return [ - FlowDisable(**flow, service=self) - for flow in flows_data - if flow["name"] not in tracked_names - ] - - def _disable_and_register(self, flow: FlowDisable) -> None: - """Disable a flow in Prefect and create its tracking record in the database. - - Args: - flow: The ``FlowDisable`` instance to disable and register. - """ - self.disable_flow_schedule(flow_id=flow.id) - DisabledFlowSchedule.objects.create(flow_name=flow.name, flow_id=flow.id) - - def _send_disable_notification( - self, flows_in_alert: list[FlowDisable], flows_disabled: list[FlowDisable] - ) -> None: - """Send a Discord notification summarising flows in alert and newly disabled flows. - - Splits the message into chunks of at most 2000 characters to respect - Discord's per-message limit. - - Args: - flows_in_alert: All untracked failing flows (warning list). - flows_disabled: Flows that were disabled in this run. - """ - message_parts = [ - self.format_flows("🚨 Flows em alerta", flows_in_alert), - self.format_flows( - f"⛔ Flows desativados <@&{Constants.DISCORD_ROLE_DADOS.value}>", - flows_disabled, - ), - ] - for chunk in self._split_message("\n\n".join(message_parts)): - send_discord_message(chunk) - - # ------------------------------------------------------------------ - # Formatting helpers - # ------------------------------------------------------------------ - - @staticmethod - def _split_message(message: str, limit: int = 2000) -> list[str]: - """Split a message into chunks that do not exceed ``limit`` characters. - - Splits on line boundaries to avoid cutting lines in the middle. - - Args: - message: The full message string to split. - limit: Maximum number of characters per chunk. Defaults to 2000. - - Returns: - List of message chunks, each at most ``limit`` characters long. - """ - chunks, current = [], "" - for line in message.splitlines(keepends=True): - if len(current) + len(line) > limit: - chunks.append(current) - current = "" - current += line - if current: - chunks.append(current) - return chunks - - @staticmethod - def format_flows(title: str, flows: list[FlowDisable]) -> str: - """Format a list of flows as a Discord markdown block. - - Args: - title: Section heading displayed in bold. - flows: List of ``FlowDisable`` instances to format. - - Returns: - Formatted string with one line per flow, or a ``_(nenhum)_`` - placeholder when the list is empty. - """ - if not flows: - return f"**{title}**\n_(nenhum)_" - - lines = [f"**{title}**"] - for flow in flows: - link = Constants.PREFECT_URL_FLOW.value + flow.id - last_run = flow.runs[0] - if last_run.task_runs: - lines.append( - Constants.TEXT_FLOW_FORMAT.value.format( - task_name=last_run.task_runs.task.name, - run_name=last_run.name, - link=link, - ) - ) - return "\n".join(lines) diff --git a/backend/apps/admin_data_tools/management/commands/disable_unhealthy_flow_schedules.py b/backend/apps/admin_data_tools/management/commands/disable_unhealthy_flow_schedules.py deleted file mode 100644 index c23babd3..00000000 --- a/backend/apps/admin_data_tools/management/commands/disable_unhealthy_flow_schedules.py +++ /dev/null @@ -1,28 +0,0 @@ -# -*- coding: utf-8 -*- -from django.core.management.base import BaseCommand - -from ._disable_unhealthy_flow_schedules.service import FlowService - - -class Command(BaseCommand): - """Management command to detect and disable unhealthy Prefect flow schedules. - - Runs the two-phase pipeline defined in ``FlowService``: - - 1. Re-enforces flows that should remain disabled but were reactivated by - Prefect (e.g. after re-registration), and detects new failures in flows - that were previously reactivated by an admin. - 2. Detects new untracked flows that have been failing, disables them in - Prefect, registers them in the database, and sends a Discord notification. - """ - - help = "Disable unhealthy flow schedules" - - def handle(self, *args, **options) -> None: - """Execute the command. - - Args: - *args: Unused positional arguments. - **options: Unused keyword arguments. - """ - FlowService().disable_unhealthy_flow_schedules() diff --git a/backend/settings/base.py b/backend/settings/base.py index 913fce85..738a6050 100644 --- a/backend/settings/base.py +++ b/backend/settings/base.py @@ -241,9 +241,9 @@ {"name": "Metadados", "app": "v1"}, {"model": "v1.dataset"}, {"model": "v1.table"}, - {"name": "Metabase", "url": "https://perguntas.basedosdados.org"}, - {"name": "Prefect", "url": "https://prefect.basedosdados.org"}, - {"name": "Grafana", "url": "https://grafana.basedosdados.org"}, + {"name": "Metabase", "url": "https://perguntas.basedosdados.org", "new_window": True}, + {"name": "Prefect", "url": "https://prefect3.basedosdados.org", "new_window": True}, + {"name": "Grafana", "url": "https://grafana.basedosdados.org", "new_window": True}, ], "show_ui_builder": True, }