diff --git a/README.md b/README.md index eb75cfd..4a239d3 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,36 @@ A high-performance, scalable workflow orchestration system built on Temporal.io for automated pipeline management, notification delivery, and resource coordination. The Orchestrator provides enterprise-grade workflow automation, real-time alerting, and comprehensive monitoring capabilities for the SIENTIA platform. +## 📑 Table of Contents + +- [Features](#features) + - [Core Functionality](#core-functionality) + - [Advanced Capabilities](#advanced-capabilities) +- [Architecture](#architecture) + - [Architecture Principles](#architecture-principles) + - [Task Queue Isolation](#2-task-queue-isolation) +- [Workflows](#-workflows) + - [Orchestrator Workflow](#1-orchestrator-workflow-orchestratorpy) + - [Alerts Workflow](#2-alerts-workflow-alertspy) + - [Reports Workflow](#3-reports-workflow-reportspy) + - [Subworkflows](#subworkflows) + - [Load Notification Package](#1-load-notification-package-load_notification_packagepy) + - [Process Notifications](#2-process-notifications-process_notificationspy) +- [Notification Filtering System](#-notification-filtering-system) +- [Prerequisites](#-prerequisites) +- [Installation](#-installation) +- [How to Run](#-how-to-run) +- [Configuration](#-configuration) +- [Monitoring and Metrics](#-monitoring-and-metrics) +- [Testing](#-testing) +- [Code Quality & Validation](#-code-quality--validation) +- [Development](#-development) +- [Troubleshooting](#-troubleshooting) +- [Performance Tuning](#-performance-tuning) +- [Contributing](#-contributing) +- [License](#-license) +- [Support](#-support) + ## Features ### Core Functionality @@ -646,6 +676,73 @@ pytest tests/activities/test_mongo_db.py pytest tests/workflows/test_orchestrator.py ``` +## 🛡️ Code Quality & Validation + +### Overview + +Since Python is not compiled, this project ships a validation workflow to catch issues early. Use the `validate.sh` script to run formatting, linting, type checks, security analysis, and tests in one command. + +### Validation Tools + +- **Ruff**: formatting and linting (fast, replaces Black/Flake8) +- **mypy**: static typing checks +- **Bandit**: security static analysis +- **pytest**: unit/integration tests with coverage + +### Tools Installation + +```bash +pip install -r requirements-dev.txt +``` + +### Complete Validation (recommended) + +```bash +./validate.sh +``` + +What `validate.sh` does: +1. Checks formatting with Ruff +2. Lints code with Ruff +3. Runs mypy type checking +4. Runs Bandit security analysis +5. Executes pytest with coverage (generates HTML report) + +Exit codes are propagated so CI can fail fast when quality gates are not met. + +### Individual Commands + +```bash +# 1) Format check +ruff format --check orchestrator/ tests/ + +# 2) Lint +ruff check orchestrator/ tests/ + +# 3) Type check +mypy orchestrator/ + +# 4) Security +bandit -r orchestrator/ -ll + +# 5) Tests with coverage +pytest tests/ --cov=orchestrator --cov-report=html +``` + +### Automatic Fixes + +```bash +# Apply formatting +ruff format orchestrator/ tests/ + +# Autofix common lint issues +ruff check --fix orchestrator/ tests/ +``` + +### Configuration + +Tooling is configured in `pyproject.toml` (lint rules, formatting, typing). Adjust thresholds and rules there as needed. + ## 🔧 Development ### Project Structure diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 0584ce6..4d186e1 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -34,12 +34,11 @@ class Formatters(BaseActivity): distribution across active ingestors, and implementing notification filtering for scheduled reports. - Key Features: - - Pipeline schedule configuration formatting (scouter, predictions_batch, minimal_retrain) + Key features: + - Pipeline schedule configuration formatting ("scouter", "predictions_batch", "minimal_retrain") - OPC slot distribution across active ingestors - - Notification filtering for comprehensive reports + - Notification filtering for comprehensive scheduled reports - Group-based report filtering with ignore list support - - Resource optimization algorithms - Configuration validation and transformation Args: @@ -69,10 +68,10 @@ class Formatters(BaseActivity): Temporal schedule configurations, organizing them by workflow type (scouter and laborious) and applying the appropriate configuration builders for each pipeline type. - Pipeline Types Supported: - - scouter: Data collection workflows with OPC tag configurations - - predictions_batch: ML prediction workflows with OPC write configurations - - minimal_retrain: Model retraining workflows with SQL query configurations + Pipeline types supported: + - "scouter": Data collection workflows with OPC tag configurations + - "predictions_batch": ML prediction workflows with OPC write configurations + - "minimal_retrain": Model retraining workflows with SQL query configurations Args: - input_data (dict[str, Any]): The input data containing @@ -80,7 +79,7 @@ class Formatters(BaseActivity): - pipelines (list[dict[str, Any]]): The schedules to process. Returns: - - dict[str, Any]: The schedule config dictionary + - dict[str, Any]: The schedule configuration dictionary keyed by namespace """ metadata = input_data.get('metadata', {}) @@ -137,7 +136,7 @@ class Formatters(BaseActivity): - active_ingestors (list[str]): The active ingestors to divide into slots. Returns: - - dict[str, Any]: The slot config dictionary + - dict[str, Any]: The slot configuration dictionary keyed by slot id (as string) """ metadata = input_data.get('metadata', {}) @@ -243,15 +242,14 @@ class Formatters(BaseActivity): metadata: dict[str, Any], ): """ - Compares the timestamps of the schedule and the current schedule to determine - which schedules need to be updated or created. + Compare new and current schedules to determine which should be updated or created. Args: - schedules (dict[str, Any]): The new schedules to compare. - current_schedules (dict[str, Any]): The existing schedules to compare against. - to_update (dict[str, Any]): Dictionary to populate with schedules that need updating. - to_create (dict[str, Any]): Dictionary to populate with schedules that need creating. - namespace (str): The namespace for the schedules. + schedules (dict[str, Any]): New schedules to compare. + current_schedules (dict[str, Any]): Existing schedules to compare against. + to_update (dict[str, Any]): Output accumulator for schedules that need updating. + to_create (dict[str, Any]): Output accumulator for schedules that need creating. + namespace (str): Namespace for the schedules being compared. metadata (dict[str, Any]): Metadata for logging purposes. """ for schedule_name, schedule in schedules.items(): @@ -285,7 +283,7 @@ class Formatters(BaseActivity): - schedule_config (dict[str, Any]): The schedule config to process. Returns: - - dict[str, Any]: The schedule config dictionary + - dict[str, Any]: A dictionary with keys 'to_update', 'to_create', and 'to_delete' """ metadata = input_data.get('metadata', {}) @@ -341,7 +339,7 @@ class Formatters(BaseActivity): - slot_config (dict[str, Any]): The slot config to process. Returns: - - dict[str, Any]: The slot config dictionary + - dict[str, Any]: A dictionary containing 'to_insert' and 'to_delete' """ metadata = input_data.get('metadata', {}) @@ -378,8 +376,8 @@ class Formatters(BaseActivity): Args: metadata (dict[str, Any]): Metadata for the notification. message (str): The success message to send. - notification_id (str): The ID of the notification. - attachment (str, optional): Optional attachment content for the notification. + notification_id (str): The ID of the notification to send. + attachment (Any | None, optional): Optional attachment content to include. """ self.send_notification( metadata=metadata, @@ -418,14 +416,14 @@ class Formatters(BaseActivity): Parses the report schedule data to extract success and error information. Args: - input_data (dict[str, Any]): The input data containing schedule reports. - Each item should have 'namespace', 'schedule_name', 'success', 'message', - and optionally 'attachment' fields. + input_data (list[dict[str, Any]]): The schedule reports. Each item must + contain 'namespace', 'schedule_name', 'success', 'message', and optionally + 'attachment'. Returns: - tuple[list[str], dict[str, Any]]: A tuple containing: - - List of successful schedule keys in format "namespace/schedule_name" - - Dictionary of error keys mapped to their error details + tuple[list[str], dict[str, Any]]: A tuple of: + - Successful schedule keys in the form "namespace/schedule_name" + - Error map keyed by the same string to error details """ success_keys = [ f'{value["namespace"]}/{value["schedule_name"]}' diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 8f1969f..546190c 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -17,13 +17,13 @@ with workflow.unsafe.imports_passed_through(): def clear_mongo_id(docs: list) -> list: """ - Remove the MongoDB internal `_id` field from the document. + Remove MongoDB internal `_id` fields from nested structures. Args: - docs (list): The document to clear. + docs (list): The list of documents or nested structures to clean. Returns: - list: The documents without the `_id` field. + list: The cleaned documents with `_id` fields removed wherever present. """ for doc in docs: if isinstance(doc, list): @@ -48,8 +48,8 @@ class MongoDB(BaseActivity): This class provides MongoDB database operations including document querying, aggregation, timestamp management, and collection management - with TTL indexes. It handles all MongoDB interactions required by - the orchestration system. + with TTL indexes. It centralizes all MongoDB interactions required by + the orchestration system lifecycle. Args: connection_string (str): MongoDB connection string @@ -73,7 +73,7 @@ class MongoDB(BaseActivity): self.client: MongoClient = MongoClient( self.connection_string, serverSelectionTimeoutMS=5000 ) - self.client.server_info() # Trigger an exception if connection fails + self.client.server_info() # Force early failure if connection is invalid self.database = self.client[self.database_name] @@ -86,7 +86,7 @@ class MongoDB(BaseActivity): def shutdown(self): """ - Shutdown the MongoDB connection and clean up resources. + Shutdown the MongoDB client and clean up resources. """ try: if self.client: @@ -98,7 +98,7 @@ class MongoDB(BaseActivity): def __del__(self): """ - Destructor to ensure MongoDB client is closed when the object is deleted. + Ensure the MongoDB client is closed when the object is garbage-collected. """ self.shutdown() @@ -111,7 +111,7 @@ class MongoDB(BaseActivity): filters (dict[str, Any]): The query filters to apply. Returns: - list[dict[str, Any]]: List of documents matching the filters, with _id fields removed. + list[dict[str, Any]]: Documents matching the filters (with `_id` removed). """ collection = self.database[collection_name] @@ -131,9 +131,10 @@ class MongoDB(BaseActivity): Args: - input_data (dict): Input data containing query parameters. Contains: - query (dict): Query parameters to filter documents. + - timestamp_fields (list[str], optional): Fields to format as RFC3339 with TZ. Returns: - list[dict]: List of documents matching the query. + list[dict]: Documents matching the query with timestamp fields normalized. """ query = input_data.get('query', {}) @@ -195,10 +196,11 @@ class MongoDB(BaseActivity): Args: - input_data (dict): Input data containing aggregation parameters. Contains: - - query (dict): Query parameters to filter documents. + - query (dict): Aggregation parameters including 'collection' and 'aggregation'. + - timestamp_fields (list[str], optional): Fields to format as RFC3339 with TZ. Returns: - list[dict]: List of aggregated documents. + list[dict]: Aggregated documents with timestamp fields normalized. """ query = input_data.get('query', {}) @@ -260,9 +262,10 @@ class MongoDB(BaseActivity): @activity.defn(name='update_pipelines_timestamps') async def update_pipelines_timestamps(self, input_data: dict[str, Any]) -> None: """ - Update the timestamps of the pipelines in the MongoDB collection. + Update `updated_at` timestamps for successfully updated pipelines. + input_data: - - updated_pipelines (list): List of updated pipelines. + - updated_pipelines (list[dict]): Pipelines with success flags to consider. """ updated_pipelines = input_data.get('updated_pipelines', []) metadata = input_data.get('metadata', {}) @@ -304,9 +307,10 @@ class MongoDB(BaseActivity): @activity.defn(name='create_pipelines_timestamps') async def create_pipelines_timestamps(self, input_data: dict[str, Any]) -> None: """ - Create the timestamps of the pipelines in the MongoDB collection. + Insert `updated_at` timestamps for newly created pipelines. + input_data: - - created_pipelines (list): List of created pipelines. + - created_pipelines (list[dict]): Pipelines with success flags to consider. """ created_pipelines = input_data.get('created_pipelines', []) metadata = input_data.get('metadata', {}) @@ -354,9 +358,10 @@ class MongoDB(BaseActivity): @activity.defn(name='delete_pipelines_timestamps') async def delete_pipelines_timestamps(self, input_data: dict[str, Any]) -> None: """ - Delete the timestamps of the pipelines in the MongoDB collection. + Delete timestamp rows for successfully deleted pipelines. + input_data: - - deleted_pipelines (list): List of deleted pipelines. + - deleted_pipelines (list[dict]): Pipelines with success flags to consider. """ deleted_pipelines = input_data.get('deleted_pipelines', []) metadata = input_data.get('metadata', {}) diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index 1de072f..71bf7e6 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -62,8 +62,8 @@ class TemporalManager(BaseActivity): async def connect_to_temporal(self): """ - Connect to Temporal server namespaces for scouter and laborious workflows. - Creates client connections to both namespaces and stores them for later use. + Connect to Temporal server namespaces used by scouter and laborious workflows. + Creates and caches `Client` connections for both namespaces for later use. """ self.logger.info(f'Connecting to Temporal side namespaces at {self.temporal_host}') self.logger.info(f'Scouter namespace: {self.scouter_namespace}') @@ -86,9 +86,13 @@ class TemporalManager(BaseActivity): @activity.defn(name='normalize_schedules') async def normalize_schedules(self, input_data: dict[str, Any]): """ - Normalize schedules. Removes schedules with no update time in mongo db collection "orchestrated_schedules". + Normalize schedules by removing orphaned schedules from Temporal. + + Any schedule marked with search attribute `orchestrated=true` that does not + exist in MongoDB collection `orchestrated_schedules` will be deleted. + input_data: - - orchestrated_schedules (dict[str, Any]): The orchestrated schedules to compare. + - orchestrated_schedules (dict[str, Any]): Current orchestrated schedules from MongoDB. """ metadata = input_data['metadata'] @@ -139,7 +143,7 @@ class TemporalManager(BaseActivity): @activity.defn(name='create_schedules') async def create_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: """ - Create schedules in Temporal + Create schedules in Temporal. Args: - input_data (dict[str, Any]): The input data containing @@ -147,7 +151,7 @@ class TemporalManager(BaseActivity): - schedules (dict[str, Any]): The schedules to create. Returns: - - dict[str, Any]: A report of the created schedules. + - list[dict[str, Any]]: Report entries for each attempted schedule creation. """ schedules_to_create = input_data['schedules'] @@ -247,7 +251,7 @@ class TemporalManager(BaseActivity): @activity.defn(name='update_schedules') async def update_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: """ - Update schedules in Temporal + Update schedules in Temporal. Args: - input_data (dict[str, Any]): The input data containing @@ -255,7 +259,7 @@ class TemporalManager(BaseActivity): - schedules (dict[str, Any]): The schedules to update. Returns: - - dict[str, Any]: A report of the updated schedules. + - list[dict[str, Any]]: Report entries for each attempted schedule update. """ schedules_to_update = input_data['schedules'] @@ -342,7 +346,7 @@ class TemporalManager(BaseActivity): @activity.defn(name='delete_schedules') async def delete_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: """ - Delete schedules in Temporal + Delete schedules in Temporal. Args: - input_data (dict[str, Any]): The input data containing @@ -350,7 +354,7 @@ class TemporalManager(BaseActivity): - schedules (list[str]): The schedules to delete. Returns: - - dict[str, Any]: A report of the deleted schedules. + - list[dict[str, Any]]: Report entries for each attempted schedule deletion. """ schedules_to_delete = input_data['schedules'] diff --git a/orchestrator/metrics.py b/orchestrator/metrics.py index cd7379e..9521a51 100644 --- a/orchestrator/metrics.py +++ b/orchestrator/metrics.py @@ -1,9 +1,8 @@ """ -Prometheus metrics definitions for the orchestrator application. +Prometheus metric definitions for the orchestrator application. -This module defines all Prometheus metrics used for monitoring the -orchestrator system including application health, email delivery, -and workflow execution metrics. +This module exposes Prometheus counters and gauges for monitoring the +orchestrator, including application health and email delivery metrics. """ from prometheus_client import Counter, Gauge @@ -19,6 +18,6 @@ CORE_LABELS = ['pod_id', 'model_name', 'pipeline_name'] EMAIL_SENT_COUNT = Counter( 'email_sent_count', - 'Number of emails sent', + 'Total number of emails sent by the orchestrator', [*CORE_LABELS, 'email_group'], ) diff --git a/orchestrator/utils/email_builder.py b/orchestrator/utils/email_builder.py index 2a815e5..db67591 100644 --- a/orchestrator/utils/email_builder.py +++ b/orchestrator/utils/email_builder.py @@ -39,7 +39,7 @@ class EmailBuilder: Returns: str: The rendered template with parameters replaced. """ - # Criar um template Jinja2 + # Create a Jinja2 template from the provided string template_obj = Template(template) return template_obj.render(parameters)