gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1222

Browse Source 
Enhance README and activities with advanced notification filtering features

- Updated README to include new intelligent notification filtering capabilities, including TTL-based duplicate prevention and ignore lists.
- Enhanced `Formatters` activity to support notification filtering for scheduled reports and added detailed filtering logic.
- Improved `SlotManager` to implement advanced notification filtering with persistent alert detection and group-based filtering.
- Updated `Alerts` workflow to incorporate intelligent filtering for real-time error notifications, ensuring efficient alert delivery.
This commit is contained in:
vitor-aignosi
2025-09-16 16:36:32 -03:00
parent 1f8d3e8777
commit b816ad523b
4 changed files with 210 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
README.md
View File

@@ -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
```

59
orchestrator/activities/formatters.py
View File

@@ -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']

66
orchestrator/activities/slot_manager.py
View File

@@ -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']

34
orchestrator/workflows/alerts.py
View File

@@ -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': {