diff --git a/README.md b/README.md index 8665909..eb75cfd 100644 --- a/README.md +++ b/README.md @@ -13,10 +13,15 @@ A high-performance, scalable workflow orchestration system built on Temporal.io ### Advanced Capabilities - **Incremental Processing**: Timestamp-based data loading to avoid reprocessing -- **Configurable Filtering**: User group-based notification filtering with custom policies +- **Intelligent Notification Filtering**: Advanced filtering system with: + - User group-based notification filtering with custom policies + - TTL-based duplicate prevention for alerts + - Ignore lists for specific notifications + - Persistent alert detection for ongoing issues - **Auto-scaling Workers**: Multiple worker instances with task queue isolation - **Comprehensive Logging**: Structured logging with PostgreSQL audit trails - **Prometheus Metrics**: Real-time monitoring and alerting integration +- **Template-based Email Generation**: Jinja2-powered HTML email templates ## Architecture @@ -252,6 +257,12 @@ flowchart LR - `notification_package` (list[dict]): Retrieved notifications - `sending_configs` (list[dict]): Active receiver group configurations +**Key Activities**: +- `get_last_data_timestamp`: Retrieves last processed timestamp from Redis +- `find_documents_in_mongodb`: Loads receiver group configurations +- `load_latest_data`: Loads notifications with timestamp filtering +- `put_last_data_timestamp`: Updates last processed timestamp + **Architecture**: ```mermaid @@ -277,10 +288,11 @@ flowchart LR **Purpose**: Handles email generation, delivery, and audit logging for notification workflows. **Key Features**: -- **HTML Generation**: Creates formatted email content for each receiver group +- **HTML Generation**: Creates formatted email content for each receiver group using Jinja2 templates - **Email Delivery**: Sends emails with attachment support and error handling - **Audit Logging**: Records delivery status and metrics in PostgreSQL - **Error Recovery**: Handles SMTP failures with detailed error reporting +- **Template Support**: Uses customizable HTML templates for different email types **Input Parameters**: ```json @@ -300,6 +312,12 @@ flowchart LR **Returns**: - `log_report` (list[dict]): Detailed delivery status for each notification +**Key Activities**: +- `build_email_html`: Generates HTML content using Jinja2 templates +- `send_email`: Delivers emails to receiver groups with error handling +- `format_log_report`: Formats delivery results for database storage +- `export_data_to_postgres`: Stores audit logs in PostgreSQL + **Architecture**: ```mermaid @@ -313,6 +331,29 @@ flowchart LR ``` +## 🔔 Notification Filtering System + +The orchestrator includes an advanced notification filtering system that prevents alert spam and ensures relevant notifications reach the appropriate user groups. + +### **Alert Filtering (`filter_notification_alerts`)** +- **Purpose**: Filters error-level notifications for immediate alerts +- **TTL Management**: Prevents duplicate alerts using configurable time-to-live settings +- **Persistent Detection**: Identifies ongoing issues that require escalation +- **Group-based Filtering**: Routes notifications to appropriate receiver groups +- **Ignore Lists**: Supports notification exclusion per group + +### **Report Filtering (`filter_notification_reports`)** +- **Purpose**: Filters all notification levels for comprehensive reports +- **Comprehensive Coverage**: Includes ERROR, WARNING, INFO, and DEBUG levels +- **Group Customization**: Applies different content policies per receiver group +- **Scheduled Processing**: Designed for regular report generation + +### **Notification Caching (`store_notification_cache`)** +- **Purpose**: Manages Redis-based notification cache for TTL enforcement +- **TTL Support**: Configurable expiration times for different notification types +- **Duplicate Prevention**: Ensures notifications aren't sent repeatedly within TTL window +- **Key Management**: Uses structured keys for efficient cache lookups + ### Key Components #### **Worker (`orchestrator/worker/worker.py`)** @@ -332,19 +373,23 @@ flowchart LR #### **Activities (`orchestrator/activities/`)** - **Activities**: Main activity orchestrator combining all operations - **TemporalManager**: Temporal schedule CRUD operations across namespaces -- **SlotManager**: Redis-based OPC slot and cache management +- **SlotManager**: Redis-based OPC slot and cache management with notification filtering - **MongoDB**: Document operations, aggregations, and TTL management - **Email**: SMTP operations with HTML generation and attachment support -- **Formatters**: Configuration processing and slot distribution algorithms +- **Formatters**: Configuration processing, slot distribution algorithms, and notification filtering for reports +- **Postgres**: PostgreSQL operations for audit logging and data export (via sientia-dataops-library) +- **Couchbase**: Database operations (currently unused but maintained for future use) #### **Utilities (`orchestrator/utils/`)** - **Connectors Configuration**: Database and service configuration management -- **Email Builder**: HTML email template generation and formatting -- **Orchestrator Functions**: Pipeline configuration transformation utilities -- **Converters**: Data type conversion and validation utilities +- **Email Builder**: HTML email template generation and formatting using Jinja2 templates +- **Orchestrator Functions**: Pipeline configuration transformation utilities for scouter, predictions_batch, and minimal_retrain workflows +- **Converters**: Data type conversion and validation utilities including frequency parsing +- **Templates**: HTML email templates for alerts and reports ## 📋 Prerequisites +### System Requirements - Python 3.11+ - Temporal server/cluster - Redis server @@ -461,6 +506,12 @@ python -m orchestrator.worker.worker | `EMAIL_SMTP_PORT` | SMTP port | `587` | No | | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No | +| `POSTGRES_MIN_CONNECTIONS` | PostgreSQL minimum connections | `10` | No | +| `POSTGRES_MAX_CONNECTIONS` | PostgreSQL maximum connections | `40` | No | +| `MONGODB_TTL_INDEX_HOURS` | MongoDB TTL index expiration in hours | `1` | No | +| `PROJECT_NAME` | Project name for metrics and logging | `sientia-orchestrator` | No | +| `LOG_LEVEL` | Application logging level | `INFO` | No | +| `KAFKA_BOOTSTRAP_SERVERS` | Kafka bootstrap servers | - | No | ### Workflow Configuration @@ -560,11 +611,28 @@ The Orchestrator system exposes comprehensive Prometheus metrics: ``` tests/ ├── orchestrator/ # Orchestrator workflow tests +│ ├── test_activities.py +│ ├── test_couchbase.py +│ ├── test_email.py +│ ├── test_formatters.py +│ ├── test_mongo_db.py +│ ├── test_slot_manager.py +│ ├── test_temporal_manager.py +│ └── test_workflows.py ├── activities/ # Activity implementation tests ├── utils/ # Utility function tests └── integration/ # End-to-end workflow tests ``` +### Test Coverage +The project maintains comprehensive test coverage including: +- **Activity Tests**: Unit tests for all activity classes +- **Workflow Tests**: Integration tests for workflow orchestration +- **Utility Tests**: Tests for configuration builders and converters +- **Database Tests**: Tests for MongoDB, Redis, and PostgreSQL operations +- **Email Tests**: Tests for email generation and delivery +- **Notification Tests**: Tests for filtering and caching logic + ### Test Execution ```bash # Install test dependencies @@ -586,21 +654,28 @@ orchestrator/ ├── activities/ # Temporal activity implementations │ ├── activities.py # Main activities orchestrator │ ├── temporal_manager.py # Temporal schedule operations -│ ├── slot_manager.py # Redis slot management +│ ├── slot_manager.py # Redis slot management and notification filtering │ ├── mongo_db.py # MongoDB operations │ ├── email.py # Email service operations -│ └── formatters.py # Configuration formatting +│ ├── formatters.py # Configuration formatting and report filtering +│ └── couchbase.py # Couchbase operations (currently unused) ├── workflows/ # Temporal workflow definitions │ ├── orchestrator.py # Main orchestration workflow │ ├── alerts.py # Error alert workflow │ ├── reports.py # Scheduled report workflow │ └── subworkflows/ # Sub-workflow implementations +│ ├── load_notification_package.py # Notification data loading +│ └── process_notifications.py # Email processing and delivery ├── worker/ # Worker implementation │ └── worker.py # Main worker orchestrator ├── utils/ # Utility functions │ ├── connectors_config.py # Database configuration │ ├── email_builder.py # Email template generation -│ └── orchestrator_functions.py # Pipeline utilities +│ ├── orchestrator_functions.py # Pipeline utilities +│ ├── converters.py # Data type conversion utilities +│ └── templates/ # Email HTML templates +│ ├── email_template.html # Report email template +│ └── general_template.html # General email template └── metrics.py # Prometheus metrics definitions ``` diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 9b0dcb8..6928bd4 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -19,12 +19,20 @@ topic_separator = "\n ========== \n" class Formatters(BaseActivity): """ - Schedule and slot configuration formatting activity. + Schedule and slot configuration formatting and notification filtering activity. - This class provides formatting operations for schedules and OPC slots, - converting pipeline configurations into Temporal-compatible formats - and managing slot distribution across active ingestors for optimal - resource utilization. + This class provides comprehensive formatting operations for schedules and OPC slots, + converting pipeline configurations into Temporal-compatible formats, managing slot + distribution across active ingestors, and implementing notification filtering for + scheduled reports. + + Key Features: + - Pipeline schedule configuration formatting (scouter, predictions_batch, minimal_retrain) + - OPC slot distribution across active ingestors + - Notification filtering for comprehensive reports + - Group-based report filtering with ignore list support + - Resource optimization algorithms + - Configuration validation and transformation Args: scouter_namespace (str): Scouter workflow namespace @@ -45,8 +53,16 @@ class Formatters(BaseActivity): @activity.defn(name="process_schedules") async def process_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Process schedules. Generates a schedule config dictionary - based on the input data workflow type. + Process pipeline configurations into Temporal-compatible schedule configurations. + + This method transforms pipeline configurations from MongoDB into properly formatted + 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 Args: - input_data (dict[str, Any]): The input data containing @@ -629,17 +645,34 @@ class Formatters(BaseActivity): @activity.defn(name="filter_notification_reports") async def filter_notification_reports(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Filters notification reports based on sending configurations and notification package. + Filter notifications for comprehensive scheduled reports. + + This method filters notifications of all levels (ERROR, WARNING, INFO, DEBUG) + for scheduled report generation. Unlike alert filtering, this method does not + implement TTL-based duplicate prevention since reports are meant to provide + comprehensive coverage of system activity within a time window. + + Filtering Logic: + - Processes all notification levels (not just ERROR) + - Applies group-specific content filtering for "reports" type + - Respects ignore lists for each receiver group + - Prevents duplicate notifications within the same report + - Groups notifications by receiver group configurations Args: input_data (dict[str, Any]): The input data containing: - - metadata (dict): Metadata for logging purposes. - - notification_package (list): The package of notifications to filter. - - sending_configs (list): The configurations for sending notifications. - Each config should have 'group_name', 'contents', and optionally 'ignore' fields. + - metadata (dict): Workflow execution metadata for logging + - notification_package (list): All notifications to filter (any level) + - sending_configs (list): Receiver group configurations with: + - group_name (str): Name of the receiver group + - contents (list): Content types to include (must contain "reports") + - ignore (list, optional): Notification IDs to exclude Returns: - dict[str, Any]: The filtered receiver groups with their notifications. + dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name. + Each group contains: + - All receiver group configuration fields + - notifications (list): Filtered notifications for this group """ metadata = input_data['metadata'] notification_package = input_data['notification_package'] diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index aea49cc..9accfd1 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -16,12 +16,20 @@ with workflow.unsafe.imports_passed_through(): class SlotManager(Redis): """ - Redis-based OPC slot management activity. + Redis-based OPC slot management and notification filtering activity. - This class manages OPC server slots and notification processing through - Redis operations. It provides functionality for loading, updating, and - deleting OPC slots, managing active ingestors, and handling notification - caching with timestamp management. + This class manages OPC server slots and provides advanced notification + filtering capabilities through Redis operations. It handles OPC slot + lifecycle management, active ingestor tracking, and implements intelligent + notification filtering with TTL-based duplicate prevention. + + Key Features: + - OPC slot loading, updating, and deletion + - Active ingestor management + - Notification filtering for alerts with TTL management + - Persistent alert detection for ongoing issues + - Notification caching with configurable expiration + - Group-based filtering with ignore list support Args: host (str): Redis server hostname @@ -42,10 +50,23 @@ class SlotManager(Redis): @activity.defn(name="load_opc_slots") async def load_opc_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Load all OPC slots from Redis + Load all OPC slots from Redis for current system state assessment. + + This method retrieves all OPC server slot configurations from Redis, + which are used to determine current resource allocation and identify + changes needed for pipeline orchestration. + + Args: + input_data (dict[str, Any]): Activity input containing metadata Returns: - dict[str, Any]: A dictionary of OPC slots + dict[str, Any]: Dictionary of OPC slots keyed by server ID, where each slot contains: + - Configuration parameters for OPC server connections + - Active pipeline assignments + - Resource allocation details + + Raises: + Exception: If Redis connection fails or data retrieval errors occur """ metadata = input_data.get("metadata", {}) @@ -313,18 +334,35 @@ class SlotManager(Redis): @activity.defn(name="filter_notification_alerts") async def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Filter notification alerts based on sending configurations and notification package. + Filter notification alerts with intelligent TTL-based duplicate prevention. + + This method implements advanced notification filtering for ERROR-level alerts, + preventing spam through TTL management and detecting persistent issues that + require escalation. It applies user group-based filtering with configurable + ignore lists and content policies. + + Filtering Logic: + - Checks Redis cache for recently sent notifications + - Identifies "core_alerts" for new notifications + - Detects "persistent_alerts" for ongoing issues beyond TTL + - Applies group-specific content filtering and ignore lists + - Prevents duplicate notifications within the same TTL window Args: input_data (dict[str, Any]): The input data containing: - - metadata (dict): Metadata for logging purposes. - - notification_package (list): The package of notifications to filter. - - sending_configs (list): The configurations for sending notifications. - Each config should have 'group_name', 'contents', and optionally 'ignore' fields. - - notification_ttl (int): Time to live for notifications in seconds. + - metadata (dict): Workflow execution metadata for logging + - notification_package (list): ERROR-level notifications to filter + - sending_configs (list): Receiver group configurations with: + - group_name (str): Name of the receiver group + - contents (list): Alert types to include (core_alerts, persistent_alerts) + - ignore (list, optional): Notification IDs to exclude + - notification_ttl (int): Seconds before considering notification persistent Returns: - dict[str, Any]: The filtered receiver groups with their notifications. + dict[str, Any]: Filtered receiver groups with their notifications, keyed by group_name. + Each group contains: + - All receiver group configuration fields + - notifications (list): Filtered notifications for this group """ metadata = input_data['metadata'] notification_package = input_data['notification_package'] diff --git a/orchestrator/workflows/alerts.py b/orchestrator/workflows/alerts.py index 90cef34..5f2ec31 100644 --- a/orchestrator/workflows/alerts.py +++ b/orchestrator/workflows/alerts.py @@ -9,22 +9,42 @@ with workflow.unsafe.imports_passed_through(): @workflow.defn(name="alerts") class Alerts: + """ + Alerts workflow for real-time error notification delivery. + + This workflow processes ERROR-level notifications from the notification queue + and sends immediate alerts to configured user groups. It implements intelligent + filtering with TTL-based duplicate prevention and persistent alert detection + for ongoing issues. + """ + @workflow.run async def run(self, input_data: dict[str, Any]): """ - Workflow to send alerts to the users + Execute the alerts workflow for real-time error notification delivery. + + This workflow loads ERROR-level notifications from MongoDB, applies + intelligent filtering with TTL management to prevent alert spam, + and sends immediate email alerts to configured receiver groups. + + The workflow implements: + - TTL-based duplicate prevention for notifications + - Persistent alert detection for ongoing issues + - User group-based filtering with ignore lists + - Audit logging of alert delivery status Args: - input_data (dict[str, Any]): Input data. It contains the following keys: - - schedule_name: str - Name of the schedule - - notification_ttl: int - Period before consider some notification persistent - - sent_ttl: int - Time to live for the sent notification + input_data (dict[str, Any]): Workflow input parameters. + Required fields: + - schedule_name (str): Name of the alert schedule + - notification_ttl (int): Seconds before considering notification persistent + - sent_ttl (int): Time-to-live for sent notification cache Returns: - None + None: Workflow completes without return value Raises: - Exception: If the workflow fails + Exception: If alert processing or delivery fails """ metadata = { 'metadata': {