diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..e94fd5f --- /dev/null +++ b/.env.example @@ -0,0 +1,36 @@ +REDIS_HOST="redis-master.redis.svc.cluster.local" +REDIS_PORT="6379" +REDIS_USERNAME="redis_username" +REDIS_PASSWORD="redis_password" + +COUCHBASE_CONNECTION_STRING="couchbase://sientia.couchbase.svc.cluster.local" +COUCHBASE_USERNAME="sientia" +COUCHBASE_PASSWORD="sientia" + +MONGODB_USERNAME="mongo_username" +MONGODB_PASSWORD="mongo_password" +MONGODB_URL="my-release-mongodb.mongodb.svc.cluster.local:27017" +MONGODB_DATABASE="sientia" +MONGODB_TTL_INDEX_HOURS="1" + +EMAIL_SENDER="aignosi@aignosi.com.br" +EMAIL_SENDER_PASSWORD="smtp_password" +EMAIL_SMTP_SERVER="smtp.gmail.com" +EMAIL_SMTP_PORT="587" + +POSTGRES_HOST="paradedb-rw.paradedb.svc.cluster.local" +POSTGRES_PORT="5432" +POSTGRES_USER="sientia" +POSTGRES_PASSWORD="sientia" +POSTGRES_DBNAME="sientia" +POSTGRES_MIN_CONNECTIONS="10" +POSTGRES_MAX_CONNECTIONS="40" + +LOG_LEVEL="DEBUG" +HTTP_METRICS_PORT="9090" +PROJECT_NAME="sientia-orchestrator" + +TEMPORAL_HOST="temporal-frontend.temporal.svc.cluster.local:7233" +TEMPORAL_NAMESPACE="default" +TEMPORAL_SCOUTER_NAMESPACE="scouter" +TEMPORAL_LABORIOUS_NAMESPACE="laborious" \ No newline at end of file diff --git a/Dockerfile b/Dockerfile deleted file mode 100644 index 95984aa..0000000 --- a/Dockerfile +++ /dev/null @@ -1,83 +0,0 @@ -FROM python:3.11-bookworm -LABEL description="Deploy Mage on ECS" -ARG FEATURE_BRANCH -USER root -SHELL ["/bin/bash", "-o", "pipefail", "-c"] - -# Definir Python 3.11 como padrΓ£o -ENV PATH="/usr/local/bin/python3.11:$PATH" -RUN update-alternatives --install /usr/bin/python python /usr/local/bin/python3.11 1 && \ - update-alternatives --install /usr/bin/python3 python3 /usr/local/bin/python3.11 1 && \ - update-alternatives --config python3 <<< '1' && \ - update-alternatives --config python <<< '1' - -## System Packages -RUN \ - curl https://packages.microsoft.com/keys/microsoft.asc | apt-key add - && \ - curl https://packages.microsoft.com/config/debian/11/prod.list > /etc/apt/sources.list.d/mssql-release.list && \ - apt-get -y update && \ - ACCEPT_EULA=Y apt-get -y install --no-install-recommends \ - # NFS dependencies - nfs-common \ - # odbc dependencies - msodbcsql18 \ - unixodbc-dev \ - graphviz \ - # postgres dependencies - postgresql-client \ - # R - r-base && \ - apt-get clean && \ - rm -rf /var/lib/apt/lists/* - -## R Packages -RUN \ - R -e "install.packages('pacman', repos='http://cran.us.r-project.org')" && \ - R -e "install.packages('renv', repos='http://cran.us.r-project.org')" - -## Python Packages -RUN \ - pip3 install --no-cache-dir sparkmagic && \ - mkdir ~/.sparkmagic && \ - curl https://raw.githubusercontent.com/jupyter-incubator/sparkmagic/master/sparkmagic/example_config.json > ~/.sparkmagic/config.json && \ - sed -i 's/localhost:8998/host.docker.internal:9999/g' ~/.sparkmagic/config.json && \ - jupyter-kernelspec install --user "$(pip3 show sparkmagic | grep Location | cut -d' ' -f2)/sparkmagic/kernels/pysparkkernel" - -# Mage integrations and other related packages -RUN \ - pip3 install --no-cache-dir "git+https://github.com/wbond/oscrypto.git@d5f3437ed24257895ae1edd9e503cfb352e635a8" && \ - pip3 install --no-cache-dir "git+https://github.com/dremio-hub/arrow-flight-client-examples.git#egg=dremio-flight&subdirectory=python/dremio-flight" && \ - pip3 install --no-cache-dir "git+https://github.com/mage-ai/singer-python.git#egg=singer-python" && \ - pip3 install --no-cache-dir "git+https://github.com/mage-ai/dbt-mysql.git#egg=dbt-mysql" && \ - pip3 install --no-cache-dir "git+https://github.com/mage-ai/sqlglot#egg=sqlglot" && \ - pip3 install --no-cache-dir faster-fifo && \ - if [ -z "$FEATURE_BRANCH" ] || [ "$FEATURE_BRANCH" = "null" ]; then \ - pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git#egg=mage-integrations&subdirectory=mage_integrations"; \ - else \ - pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git@$FEATURE_BRANCH#egg=mage-integrations&subdirectory=mage_integrations"; \ - fi - -# Mage -COPY ./mage_ai/server/constants.py /tmp/constants.py -RUN if [ -z "$FEATURE_BRANCH" ] || [ "$FEATURE_BRANCH" = "null" ] ; then \ - tag=$(tail -n 1 /tmp/constants.py) && \ - VERSION=$(echo "$tag" | tr -d "'") && \ - pip3 install --no-cache-dir "mage-ai[all]==$VERSION"; \ - else \ - pip3 install --no-cache-dir "git+https://github.com/mage-ai/mage-ai.git@$FEATURE_BRANCH#egg=mage-ai[all]"; \ - fi - -## Startup Script -COPY --chmod=0755 ./scripts/install_other_dependencies.py ./scripts/run_app.sh /app/ -ENV MAGE_DATA_DIR="/home/src/mage_data" -ENV PYTHONPATH="${PYTHONPATH}:/home/src" -WORKDIR /home/src -EXPOSE 6789 -EXPOSE 7789 - -# Copia o arquivo requirements.txt para o contΓͺiner -COPY requirements.txt /app/requirements.txt -RUN pip3 install --no-cache-dir -r /app/requirements.txt - - -CMD ["/bin/sh", "-c", "/app/run_app.sh"] \ No newline at end of file diff --git a/Makefile b/Makefile deleted file mode 100644 index 66aae81..0000000 --- a/Makefile +++ /dev/null @@ -1,7 +0,0 @@ -VERSION = 1.0.8 -name = sientia-laborious -# ENVIRONMENT = production - -docker-hub: - @docker build --no-cache -t aignosi.azurecr.io/$(name):$(VERSION) . - @docker push aignosi.azurecr.io/$(name):$(VERSION) \ No newline at end of file diff --git a/README.md b/README.md index 160ec29..8665909 100644 --- a/README.md +++ b/README.md @@ -1,190 +1,692 @@ # SIENTIA DataOps Orchestrator Temporal -## Overview +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. -The SIENTIA DataOps Orchestrator Temporal is a comprehensive workflow orchestration system built on Temporal.io that manages data pipelines, notifications, and system orchestration for the SIENTIA platform. It provides automated scheduling, monitoring, and execution of data processing workflows with integrated alerting and reporting capabilities. +## Features -## Project Goals +### Core Functionality +- **Pipeline Orchestration**: Automated deployment and management of data processing pipelines +- **Real-time Notifications**: Intelligent alert filtering and delivery with TTL management +- **Resource Management**: Dynamic OPC server slot allocation and active ingestor monitoring +- **Schedule Management**: Temporal-based workflow scheduling with automatic retry policies +- **Multi-namespace Support**: Separate workflow queues for scouter and laborious operations -- **Pipeline Orchestration**: Automate the deployment and management of data processing pipelines -- **Notification Management**: Handle real-time alerts and scheduled reports for system events -- **Resource Management**: Manage OPC server slots and data ingestion resources -- **Workflow Automation**: Coordinate complex workflows across multiple services and databases -- **Monitoring & Reporting**: Provide comprehensive logging and metrics for system health +### Advanced Capabilities +- **Incremental Processing**: Timestamp-based data loading to avoid reprocessing +- **Configurable Filtering**: User group-based notification filtering with custom policies +- **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 ## Architecture -The system is built around three main worker queues, each handling specific types of workflows: +The SIENTIA DataOps Orchestrator uses a Temporal-based workflow architecture with clear separation of concerns and robust error handling. The architecture is designed for high availability, scalability, and operational excellence in data pipeline environments. -### 1. Orchestrator Queue (`orchestrator-queue`) -Handles pipeline orchestration and resource management workflows. +### Architecture Principles -### 2. Alerts Queue (`alerts-queue`) -Manages real-time alert notifications and error reporting. +#### 1. **Separation of Concerns** +- **Worker Layer**: Manages Temporal workers, task queues, and application lifecycle +- **Workflow Layer**: Orchestrates business logic and process coordination +- **Activity Layer**: Implements specific operations and external system interactions +- **Data Layer**: Handles data persistence, caching, and external service connections -### 3. Reports Queue (`reports-queue`) -Handles scheduled reports and data summaries. +#### 2. **Task Queue Isolation** +- **Orchestrator Queue**: Pipeline and resource management workflows +- **Alerts Queue**: Real-time error notification workflows +- **Reports Queue**: Scheduled reporting and summary workflows -## Workflows +#### 3. **Fault Tolerance & Resilience** +- **Automatic Retry Policies**: Configurable retry strategies for transient failures +- **Graceful Degradation**: System continues operating with reduced functionality +- **Comprehensive Error Handling**: Detailed error reporting and notification integration +- **Connection Management**: Automatic reconnection for SMTP and database services + +#### 4. **Scalability & Performance** +- **Horizontal Scaling**: Multiple worker instances for load distribution +- **Connection Pooling**: Optimized database and Redis connections +- **Asynchronous Processing**: Non-blocking operations for improved throughput +- **Resource Optimization**: Intelligent slot allocation and ingestor management + +## πŸ”„ Workflows ### Main Workflows -#### 1. Orchestrator Workflow -**Purpose**: Main orchestration workflow that manages pipeline deployment and resource allocation. +#### 1. Orchestrator Workflow (`orchestrator.py`) + +The **Orchestrator** workflow is the main coordination workflow that manages pipeline deployment and resource allocation across the SIENTIA platform. + +**Purpose**: +- **Pipeline Management**: Coordinates deployment of scouter and laborious pipelines +- **Resource Allocation**: Manages OPC server slots and active ingestor distribution +- **Schedule Synchronization**: Ensures Temporal schedules match MongoDB configurations +- **Infrastructure Management**: Creates, updates, and deletes workflow schedules + +**Execution Flow**: +1. **Configuration Loading**: Retrieves pipeline and OPC server configurations from MongoDB +2. **Resource Assessment**: Loads current OPC slots and active ingestors from Redis +3. **Schedule Processing**: Formats configurations for different workflow types +4. **Deployment Operations**: Creates, updates, or deletes Temporal schedules +5. **Resource Updates**: Updates OPC slots and MongoDB timestamps +6. **Reporting**: Generates comprehensive orchestration reports **Input Parameters**: -- `schedule_name` (str): Name of the orchestration schedule -- `pipelines_query` (dict): MongoDB query to retrieve pipeline configurations -- `opc_servers_query` (dict): MongoDB query to retrieve OPC server configurations +```json +{ + "schedule_name": "hourly_orchestration", + "pipelines_query": { + "collection": "pipelines", + "aggregation": [ + {"$match": {"active": true}}, + {"$sort": {"updated_at": -1}} + ] + }, + "opc_servers_query": { + "collection": "opc_servers", + "filters": {"active": true} + } +} +``` -**What it does**: -- Retrieves pipeline configurations from MongoDB -- Loads current OPC server slots and active ingestors from Redis -- Processes schedules and creates slot configurations -- Deploys schedules to Temporal server (scouter and laborious namespaces) -- Updates OPC slots in Redis -- Generates orchestration reports +#### Architecture -#### 2. Alerts Workflow -**Purpose**: Sends real-time error alerts to configured user groups. +```mermaid +flowchart LR + subgraph "Parallel Data Loading" + A[1. aggregate_documents_in_mongodb] + B[2. find_documents_in_mongodb
OPC Servers] + C[3. find_documents_in_mongodb
Orchestrated Schedules] + D[4. load_opc_slots] + E[5. load_active_ingestors] + end + + subgraph "Parallel Processing" + F[6. format_schedule_config] + G[7. process_schedules] + H[8. process_slots] + end + + subgraph "Parallel Config Creation" + I[9. create_schedule_config] + J[10. create_slot_config] + K[11. normalize_schedules] + L[12. create_collection_with_ttl_index] + end + + subgraph "Parallel Operations" + M[13. delete_slots] + N[14. update_slots] + O[15. delete_schedules] + P[16. create_schedules] + Q[17. update_schedules] + end + + subgraph "Parallel Reports & Timestamps" + R[18. report_schedule_orchestration] + S[19. report_slot_orchestration] + T[20. update_pipelines_timestamps] + U[21. create_pipelines_timestamps] + V[22. delete_pipelines_timestamps] + end + + A --> F + F --> I + I --> M + M --> R + + A -.-> MongoDB1[(MongoDB)] + D -.-> Redis1[(Redis)] + K -.-> Temporal[(Temporal)] + R -.-> Reports[Reports] +``` + +Note: Green blocks (🟩) indicate parallel processing operations that run concurrently for improved performance. + + +#### 2. Alerts Workflow (`alerts.py`) + +The **Alerts** workflow processes and sends real-time error notifications to configured user groups with intelligent filtering and duplicate prevention. + +**Purpose**: +- **Error Alerting**: Immediate notification of ERROR-level events +- **TTL Management**: Prevents alert spam using configurable time-to-live settings +- **Group Filtering**: Sends alerts only to relevant user groups +- **Persistent Monitoring**: Tracks and escalates persistent issues + +**Execution Flow**: +1. **Notification Loading**: Retrieves ERROR-level notifications from MongoDB +2. **Timestamp Filtering**: Applies incremental processing using Redis timestamps +3. **Group Filtering**: Filters notifications by user group configurations +4. **TTL Processing**: Checks notification cache to prevent duplicate alerts +5. **Email Generation**: Creates HTML email content for each group +6. **Delivery & Logging**: Sends emails and logs results to PostgreSQL **Input Parameters**: -- `schedule_name` (str): Name of the alert schedule -- `notification_ttl` (int): Time period before considering notifications persistent -- `sent_ttl` (int): Time to live for sent notification cache +```json +{ + "schedule_name": "error_alerts", + "notification_ttl": 3600, + "sent_ttl": 7200 +} +``` -**What it does**: -- Filters notifications by ERROR level -- Loads notification packages from MongoDB -- Applies user group filtering and notification TTL rules -- Sends HTML email alerts -- Stores notification logs in PostgreSQL -- Caches sent notifications to prevent duplicates +#### Architecture -#### 3. Reports Workflow -**Purpose**: Sends scheduled reports to configured user groups. +```mermaid +flowchart LR + A[1. load_notification_packageπŸ”ƒ] --> B[2. filter_notification_alerts] --> C[3. process_notificationsπŸ”ƒ] + + A -.-> MongoDB[(MongoDB)] + A -.-> Redis[(Redis)] + B -.-> Filters[Report Filters] + C -.-> Email[(Email)] + C -.-> PostgreSQL[(PostgreSQL)] + + style A fill:#000,color:#fff + style C fill:#000,color:#fff +``` + +#### 3. Reports Workflow (`reports.py`) + +The **Reports** workflow generates and sends scheduled comprehensive reports to configured user groups. + +**Purpose**: +- **Scheduled Reporting**: Regular summary reports of system activity +- **Comprehensive Coverage**: Includes all notification levels (not just errors) +- **Group Management**: Customizable reports per user group +- **Audit Trail**: Complete logging of report delivery + +**Execution Flow**: +1. **Data Collection**: Loads all notifications from MongoDB (any level) +2. **Timestamp Processing**: Uses incremental loading with Redis timestamps +3. **Group Processing**: Applies user group filtering for report customization +4. **Report Generation**: Creates HTML reports with comprehensive summaries +5. **Distribution**: Sends reports to configured recipients +6. **Audit Logging**: Records delivery status in PostgreSQL **Input Parameters**: -- `schedule_name` (str): Name of the report schedule +```json +{ + "schedule_name": "reports", +} +``` -**What it does**: -- Loads all notifications (any level) from MongoDB -- Applies user group filtering -- Generates HTML report emails -- Stores report logs in PostgreSQL +#### Architecture + +```mermaid +flowchart LR + A[1. load_notification_packageπŸ”ƒ] --> B[2. filter_notification_reports] --> C[3. process_notificationsπŸ”ƒ] + + A -.-> MongoDB[(MongoDB)] + A -.-> Redis[(Redis)] + B -.-> Filters[Report Filters] + C -.-> Email[(Email)] + C -.-> PostgreSQL[(PostgreSQL)] + + style A fill:#000,color:#fff + style C fill:#000,color:#fff +``` ### Subworkflows -#### 1. Load Notification Package -**Purpose**: Loads notification data and configuration from various sources. +#### 1. Load Notification Package (`load_notification_package.py`) + +**Purpose**: Centralized notification data loading and configuration management for both alerts and reports workflows. + +**Key Features**: +- **Incremental Processing**: Uses Redis timestamps for efficient data loading +- **Configuration Management**: Loads active receiver group configurations +- **Data Validation**: Ensures complete data packages before processing +- **Timestamp Management**: Updates last processed timestamps **Input Parameters**: -- `metadata` (dict): Workflow metadata -- `mail_type` (str): Type of mail (Alerts/Reports) -- `base_data_filter` (dict): Base filters for data retrieval +```json +{ + "metadata": {"workflow_name": "alerts", "schedule_name": "error_alerts"}, + "mail_type": "Alerts", + "base_data_filter": {"level": "ERROR"} +} +``` **Returns**: -- `last_timestamp` (str): Last processed timestamp -- `notification_package` (list): Package of notifications to process -- `sending_configs` (list): Email sending configurations +- `last_timestamp` (str | None): Last processed timestamp +- `notification_package` (list[dict]): Retrieved notifications +- `sending_configs` (list[dict]): Active receiver group configurations -#### 2. Process Notifications -**Purpose**: Processes notifications and sends emails with logging. +**Architecture**: + +```mermaid +flowchart LR + subgraph "Parallel Loading" + A[1. get_last_data_timestamp] + B[2. find_documents_in_mongodb
Receiver Groups] + end + + C[3. load_latest_data] --> D[4. put_last_data_timestamp] + + A --> C + B --> D + + A -.-> Redis1[(Redis)] + B -.-> MongoDB1[(MongoDB)] + C -.-> MongoDB2[(MongoDB)] + D -.-> Redis2[(Redis)] +``` + +#### 2. Process Notifications (`process_notifications.py`) + +**Purpose**: Handles email generation, delivery, and audit logging for notification workflows. + +**Key Features**: +- **HTML Generation**: Creates formatted email content for each receiver group +- **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 **Input Parameters**: -- `metadata` (dict): Workflow metadata -- `mail_type` (str): Type of mail being sent -- `schema` (str): Database schema name -- `table_name` (str): Database table name -- `notification_package` (list): Notifications to process +```json +{ + "metadata": {"workflow_name": "alerts", "schedule_name": "error_alerts"}, + "mail_type": "Alerts", + "schema": "sientia_data", + "table_name": "log_report", + "notification_package": { + "group_name": "admin_team", + "members": ["admin@example.com"], + "notifications": [...] + } +} +``` **Returns**: -- `log_report` (dict): Report of processed notifications +- `log_report` (list[dict]): Detailed delivery status for each notification -## Environment Variables +**Architecture**: -### Database Connections +```mermaid +flowchart LR + A[1. build_email_html] --> B[2. send_email] --> C[3. format_log_report] --> D[4. export_data_to_postgres] + + A -.-> HTML[HTML Generator] + B -.-> SMTP[(SMTP)] + C -.-> Formatter[Log Formatter] + D -.-> PostgreSQL[(PostgreSQL)] -#### Redis Configuration -- `REDIS_HOST`: Redis server hostname (default: localhost) -- `REDIS_PORT`: Redis server port (default: 6379) -- `REDIS_USERNAME`: Redis username (default: default) -- `REDIS_PASSWORD`: Redis password (from secret) - -#### MongoDB Configuration -- `MONGODB_USERNAME`: MongoDB username (default: root) -- `MONGODB_PASSWORD`: MongoDB password -- `MONGODB_URL`: MongoDB server URL (default: localhost:27017) -- `MONGODB_DATABASE`: Database name (default: sientia) -- `MONGODB_TTL_INDEX_HOURS`: TTL index duration in hours (default: 1) - -#### PostgreSQL Configuration -- `POSTGRES_HOST`: PostgreSQL server hostname -- `POSTGRES_PORT`: PostgreSQL server port (default: 5432) -- `POSTGRES_USER`: Database username (default: sientia) -- `POSTGRES_PASSWORD`: Database password (default: sientia) -- `POSTGRES_DBNAME`: Database name (default: sientia) -- `POSTGRES_MIN_CONNECTIONS`: Minimum connection pool size (default: 10) -- `POSTGRES_MAX_CONNECTIONS`: Maximum connection pool size (default: 40) - -#### Couchbase Configuration -- `COUCHBASE_CONNECTION_STRING`: Couchbase server connection string -- `COUCHBASE_USERNAME`: Couchbase username (default: sientia) -- `COUCHBASE_PASSWORD`: Couchbase password (default: sientia) - -### Email Configuration -- `EMAIL_SENDER`: Sender email address -- `EMAIL_SENDER_PASSWORD`: App password for SMTP authentication -- `EMAIL_SMTP_SERVER`: SMTP server hostname (default: smtp.gmail.com) -- `EMAIL_SMTP_PORT`: SMTP server port (default: 587) - -### Temporal Configuration -- `TEMPORAL_HOST`: Temporal server hostname and port -- `TEMPORAL_NAMESPACE`: Default Temporal namespace (default: default) -- `TEMPORAL_SCOUTER_NAMESPACE`: Scouter workflow namespace (default: scouter) -- `TEMPORAL_LABORIOUS_NAMESPACE`: Laborious workflow namespace (default: laborious) - -### Application Configuration -- `LOG_LEVEL`: Logging level (default: DEBUG) -- `HTTP_METRICS_PORT`: Prometheus metrics port (default: 9090) -- `PROJECT_NAME`: Project identifier (default: sientia-orchestrator) -- `POD_ID`: Kubernetes pod identifier for metrics - -### Kafka Configuration -- `KAFKA_BOOTSTRAP_SERVERS`: Kafka bootstrap servers - -## Deployment - -The system is designed for Kubernetes deployment using Helm charts with: -- Health checks and readiness probes -- Prometheus metrics endpoint -- ServiceMonitor integration for Prometheus Operator -- Configurable resource limits and scaling -- SSH key management for Git operations - -## Usage - -Workflows can be triggered via Temporal client calls with appropriate input parameters. The system automatically handles: -- Pipeline configuration retrieval -- Resource allocation -- Schedule deployment -- Notification processing -- Email delivery -- Logging and monitoring - -## Monitoring - -The system exposes Prometheus metrics at `/metrics` endpoint including: -- Application status (up/down) -- Email sent counts -- Workflow execution metrics -- Custom business metrics - -All operations are logged with structured metadata for debugging and auditing purposes. - -# PR shortcut ``` -git log origin/main..HEAD --no-merges > git_log + +### Key Components + +#### **Worker (`orchestrator/worker/worker.py`)** +- **Purpose**: Main application orchestrator managing Temporal workers and task queues +- **Responsibilities**: + - Temporal client initialization and connection management + - Worker lifecycle management and graceful shutdown + - Task queue configuration (orchestrator, alerts, reports) + - Prometheus metrics server initialization + - Notification handler setup and configuration +- **Key Features**: + - Multi-queue worker management with automatic scaling + - Health check endpoints for Kubernetes liveness/readiness probes + - Graceful shutdown with cleanup procedures + - Comprehensive error handling and metrics collection + +#### **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 +- **MongoDB**: Document operations, aggregations, and TTL management +- **Email**: SMTP operations with HTML generation and attachment support +- **Formatters**: Configuration processing and slot distribution algorithms + +#### **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 + +## πŸ“‹ Prerequisites + +- Python 3.11+ +- Temporal server/cluster +- Redis server +- MongoDB server +- PostgreSQL database +- SMTP server access + +**Note**: External dependencies must be available either through: +- Kubernetes cluster deployment +- Docker Compose setup +- Cloud-managed services +- Local installations + +## πŸš€ Installation + +### Local Development Setup + +1. **Clone the repository** + ```bash + git clone + cd sientia-dataops-orchestrator_temporal + ``` + +2. **Create virtual environment** + ```bash + python3.11 -m venv venv + source ./venv/bin/activate + ``` + +3. **Install dependencies** + ```bash + pip install -r requirements.txt + ``` + +4. **Configure environment variables** + ```bash + # Set required environment variables for services + export TEMPORAL_HOST=localhost:7233 + export REDIS_HOST=localhost + export MONGODB_URL=localhost:27017 + export POSTGRES_HOST=localhost + # ... additional configuration + ``` + +## πŸ“¦ How to Run + +### Running the Orchestrator Application + +Use the provided script to run the application locally: + +```bash +# Make script executable (first time only) +chmod +x run_local.sh + +# Run the application +./run_local.sh ``` -Prompt: -Write a summary of PR changes in markdown. Be objective and direct. Write to file + +The script will: +- Activate the virtual environment +- Load environment variables from `.env` +- Start the orchestrator worker application + +### Running Tests and Coverage + +Use the provided script to run tests with coverage: + +```bash +# Make script executable (first time only) +chmod +x run_coverage.sh + +# Run tests with coverage +./run_coverage.sh +``` + +### Manual Application Execution + +For manual execution without scripts: + +```bash +# Activate virtual environment +source ./venv/bin/activate + +# Start the orchestrator worker +python -m orchestrator.worker.worker +``` + +## βš™οΈ Configuration + +### Environment Variables + +| Variable | Description | Default | Required | +|----------|-------------|---------|----------| +| `TEMPORAL_HOST` | Temporal server address | `localhost:7233` | Yes | +| `TEMPORAL_NAMESPACE` | Default Temporal namespace | `default` | No | +| `TEMPORAL_SCOUTER_NAMESPACE` | Scouter workflow namespace | `scouter` | No | +| `TEMPORAL_LABORIOUS_NAMESPACE` | Laborious workflow namespace | `laborious` | No | +| `REDIS_HOST` | Redis server hostname | `localhost` | Yes | +| `REDIS_PORT` | Redis server port | `6379` | Yes | +| `REDIS_USERNAME` | Redis username | `default` | Yes | +| `REDIS_PASSWORD` | Redis password | - | Yes | +| `MONGODB_URL` | MongoDB server URL | `localhost:27017` | Yes | +| `MONGODB_USERNAME` | MongoDB username | `root` | Yes | +| `MONGODB_PASSWORD` | MongoDB password | - | Yes | +| `MONGODB_DATABASE_NAME` | MongoDB database name | `sientia` | Yes | +| `POSTGRES_HOST` | PostgreSQL hostname | `localhost` | Yes | +| `POSTGRES_PORT` | PostgreSQL port | `5432` | Yes | +| `POSTGRES_USER` | PostgreSQL username | `sientia` | Yes | +| `POSTGRES_PASSWORD` | PostgreSQL password | - | Yes | +| `POSTGRES_DBNAME` | PostgreSQL database | `sientia` | Yes | +| `EMAIL_SENDER` | Sender email address | - | Yes | +| `EMAIL_SENDER_PASSWORD` | SMTP password | - | Yes | +| `EMAIL_SMTP_SERVER` | SMTP server | `smtp.gmail.com` | No | +| `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 | + +### Workflow Configuration + +Temporal input configuration sample: + +#### Orchestrator Workflow + +```json +{ + "schedule_name": "orchestrator-test", + "pipelines_query": { + "collection": "pipelines", + "aggregation": [ + { + "$lookup": { + "from": "models", + "localField": "model_id", + "foreignField": "id", + "as": "model_docs" + } + }, + { + "$match": { + "active": True + } + }, + { + "$addFields": { + "models": { + "$arrayElemAt": [ + "$model_docs", + 0 + ] + } + } + }, + { + "$match": { + "models.active": True + } + }, + { + "$project": { + "model_docs": 0 + } + } + ] + }, + "opc_servers_query": { + "collection": "opc-servers", + "filters": { + + } + } +} +``` + +#### Alerts Workflow + +```json +{ + "schedule_name": "alerts", + "notification_ttl": 5*60, + "sent_ttl": 10*60 +} +``` + +#### Reports Workflow + +```json +{ + "schedule_name": "reports", +} +``` + +## πŸ“Š Monitoring and Metrics + +The Orchestrator system exposes comprehensive Prometheus metrics: + +### Application Metrics +- `app_up`: Application health status (1=healthy, 0=unhealthy) +- `email_sent_count`: Email delivery operation count by group + +### Workflow Metrics +- Schedule creation, update, and deletion success rates +- Notification processing times and error rates +- Resource allocation and slot management metrics + +### Database Metrics +- MongoDB query performance and connection health +- Redis operation counts and response times +- PostgreSQL export operations and audit log metrics + +## πŸ§ͺ Testing + +### Test Structure +``` +tests/ +β”œβ”€β”€ orchestrator/ # Orchestrator workflow tests +β”œβ”€β”€ activities/ # Activity implementation tests +β”œβ”€β”€ utils/ # Utility function tests +└── integration/ # End-to-end workflow tests +``` + +### Test Execution +```bash +# Install test dependencies +pip install pytest pytest-cov pytest-asyncio + +# Run tests with coverage +pytest --cov=orchestrator --cov-report=html + +# Run specific test modules +pytest tests/activities/test_mongo_db.py +pytest tests/workflows/test_orchestrator.py +``` + +## πŸ”§ Development + +### Project Structure +``` +orchestrator/ +β”œβ”€β”€ activities/ # Temporal activity implementations +β”‚ β”œβ”€β”€ activities.py # Main activities orchestrator +β”‚ β”œβ”€β”€ temporal_manager.py # Temporal schedule operations +β”‚ β”œβ”€β”€ slot_manager.py # Redis slot management +β”‚ β”œβ”€β”€ mongo_db.py # MongoDB operations +β”‚ β”œβ”€β”€ email.py # Email service operations +β”‚ └── formatters.py # Configuration formatting +β”œβ”€β”€ workflows/ # Temporal workflow definitions +β”‚ β”œβ”€β”€ orchestrator.py # Main orchestration workflow +β”‚ β”œβ”€β”€ alerts.py # Error alert workflow +β”‚ β”œβ”€β”€ reports.py # Scheduled report workflow +β”‚ └── subworkflows/ # Sub-workflow implementations +β”œβ”€β”€ 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 +└── metrics.py # Prometheus metrics definitions +``` + +### Adding New Features + +1. **Follow Temporal patterns** for new workflows and activities +2. **Add comprehensive docstrings** for all public methods +3. **Include Prometheus metrics** for monitoring +4. **Add unit tests** for new functionality +5. **Update this README** with new features and configuration + +## πŸ› Troubleshooting + +### Common Issues + +1. **Temporal Connection Failures** + - Verify Temporal server is running and accessible + - Check namespace configuration and permissions + - Review server logs for connection issues + +2. **Database Connection Issues** + - Verify all database services are running + - Check connection credentials and network access + - Ensure proper connection pool configuration + +3. **Email Delivery Failures** + - Verify SMTP server configuration and credentials + - Check email sender permissions and authentication + - Review email delivery logs for specific errors + +4. **Workflow Execution Failures** + - Review activity error logs and notifications + - Check MongoDB collection configurations + - Verify input data format and required fields + +### Debug Mode + +Enable debug logging by setting the log level: +```bash +export LOG_LEVEL=DEBUG +``` + +## ⚑ Performance Tuning + +### Key Parameters + +- **Worker Concurrency**: Configure worker task limits in Temporal client +- **Connection Pools**: Optimize database connection pool sizes +- **Redis TTL**: Adjust cache TTL settings based on requirements +- **Batch Sizes**: Configure notification processing batch sizes + +### Scaling Considerations + +- **Horizontal Scaling**: Deploy multiple worker instances +- **Task Queue Distribution**: Use dedicated queues for different workflows +- **Database Performance**: Optimize indexes and connection pooling +- **Memory Management**: Monitor and configure appropriate resource limits + +## 🀝 Contributing + +1. Fork the repository +2. Create a feature branch +3. Make your changes with comprehensive testing +4. Update documentation and docstrings +5. Submit a pull request + +### Code Quality Standards + +- Follow PEP 8 style guidelines +- Include comprehensive docstrings for all public methods +- Maintain test coverage above 80% +- Use type hints where appropriate +- Follow Temporal.io best practices + +## πŸ“„ License + +This project is licensed under the terms specified in the LICENSE file. + +## πŸ†˜ Support + +For support and questions: +- Check the troubleshooting section above +- Review the metrics and logs for error patterns +- Open an issue in the project repository +- Contact the development team + +--- + +**Note**: The SIENTIA DataOps Orchestrator is designed for production use in enterprise data environments. Ensure proper security configuration and network isolation for production deployments. diff --git a/coverage.sh b/coverage.sh deleted file mode 100755 index c3235b5..0000000 --- a/coverage.sh +++ /dev/null @@ -1 +0,0 @@ -pytest --cov=orchestrator --cov-report=html && xdg-open htmlcov/index.html \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml deleted file mode 100644 index e3aba82..0000000 --- a/docker-compose.yml +++ /dev/null @@ -1,148 +0,0 @@ -version: '3.8' - -services: - postgres: - image: postgres:15 - container_name: postgres - environment: - POSTGRES_USER: sientia - POSTGRES_PASSWORD: sientia - POSTGRES_DB: sientia - ports: - - "5432:5432" - volumes: - - ./postgres_data:/var/lib/postgresql/data - networks: - - sientia-network - - - - couchbase: - image: couchbase/server:7.2.0 - container_name: couchbase - ports: - - "8091:8091" # Admin UI and REST API - - "8092:8092" # Query Service (N1QL) - - "8093:8093" # Index Service - - "8094:8094" # Search Service - - "11210:11210" # Data Service (KV) - - "18091:18091" # Analytics Service (if enabled) - environment: - CB_CLUSTER_USERNAME: sientia - CB_CLUSTER_PASSWORD: sientia - CB_CLUSTER_RAMSIZE: 256 - CB_CLUSTER_INDEX_RAMSIZE: 256 - volumes: - - ./couchbase_data:/opt/couchbase/var - networks: - - sientia-network - healthcheck: - test: ["CMD-SHELL", "curl -f http://localhost:8091/pools/default || exit 1"] - interval: 10s - timeout: 10s - retries: 5 - - redis: - image: redis:7-alpine # Using a lightweight Redis image - container_name: redis - ports: - - "6379:6379" - volumes: - - ./redis_data:/data # Persist Redis data - networks: - - sientia-network - - redis-commander: - image: rediscommander/redis-commander:latest - container_name: redis-commander - environment: - REDIS_HOSTS: local:redis:6379 # Connects to the 'redis' service within the Docker network - ports: - - "8081:8081" # Access the Redis Commander UI on this port - depends_on: - - redis # Ensures Redis starts before Redis Commander - networks: - - sientia-network - - kafka: - image: bitnami/kafka:3.7 # Using a specific Kafka version for stability - container_name: kafka - ports: - - "9092:9092" # For clients connecting from the host machine or outside Docker network - environment: - # KRaft (Kafka Raft without Zookeeper) settings - KAFKA_CFG_NODE_ID: '0' - KAFKA_CFG_PROCESS_ROLES: 'broker,controller' - KAFKA_CFG_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' - # Listeners: ://: - # PLAINTEXT_EXTERNAL for host access, INTERNAL for container-to-container communication - KAFKA_CFG_LISTENERS: 'PLAINTEXT_EXTERNAL://0.0.0.0:9092,INTERNAL://0.0.0.0:19092,CONTROLLER://0.0.0.0:9093' - # Advertised Listeners: How clients (including Kafka-UI) will connect - KAFKA_CFG_ADVERTISED_LISTENERS: 'PLAINTEXT_EXTERNAL://localhost:9092,INTERNAL://kafka:19092' - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT_EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT' - KAFKA_CFG_INTER_BROKER_LISTENER_NAME: 'INTERNAL' - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: '0@kafka:9093' # Node 0 is at kafka:9093 for controller comms - - # Single node cluster settings (important for KRaft single node) - KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: '1' - KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: '1' - KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: '1' - KAFKA_CFG_DEFAULT_REPLICATION_FACTOR: '1' # For auto-created topics - KAFKA_CFG_NUM_PARTITIONS: '1' # Default partitions for auto-created topics - - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: 'true' # Convenient for development - volumes: - - kafka_data:/bitnami/kafka # Bitnami Kafka data directory - networks: - - sientia-network - healthcheck: - # Checks if Kafka is ready by trying to list topics using the internal listener - test: ["CMD-SHELL", "/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server kafka:19092 --list > /dev/null || exit 1"] - interval: 30s - timeout: 10s - retries: 5 - - kafka-ui: - image: provectuslabs/kafka-ui:latest - container_name: kafka-ui - ports: - - "8082:8080" # Kafka UI will be accessible on host's port 8082 - environment: - KAFKA_CLUSTERS_0_NAME: sientia-local-kafka - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:19092 # Connects to Kafka's internal listener - # DYNAMIC_CONFIG_ENABLED: 'true' # Optional: To allow config changes through UI - depends_on: - kafka: # Ensures Kafka starts and is healthy before Kafka UI - condition: service_healthy - networks: - - sientia-network - - mongodb: - image: mongo:7.0 - container_name: mongodb - ports: - - "27017:27017" - environment: - MONGO_INITDB_ROOT_USERNAME: sientia - MONGO_INITDB_ROOT_PASSWORD: sientia - volumes: - - mongodb_data:/data/db - networks: - - sientia-network - - -networks: - sientia-network: - driver: bridge - -volumes: - postgres_data: - driver: local - couchbase_data: - driver: local - redis_data: - driver: local - kafka_data: - driver: local - mongodb_data: - driver: local \ No newline at end of file diff --git a/email.html b/email.html deleted file mode 100644 index 5e90b26..0000000 --- a/email.html +++ /dev/null @@ -1,267 +0,0 @@ - - - - - - SIENTIAβ„’ Report - - - -

SIENTIAβ„’ Alerts

- -

Errors detected:

- -

Model: ipsum et

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
TAG_node34:tag_8_LISTENNING_STOPPEDadipiscing_eiusmod_dodolor_labore_consectetur_ipsum2025-07-29 14:39:52.952316+00:00sed elit dolore incididunt incididunt aliqua
- -

Model: incididunt et do

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
OPC_LISTENNING_STOPPED__server_5eiusmod_adipiscing_dolore_ipsum_incididunt_incididuntaliqua_dolore2025-07-29 14:39:52.952579+00:00dolor elit do ipsum consectetur amet ut do amet et
- -

Model: magna adipiscing aliqua

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERdo_dolore_ut_ametet_eiusmod2025-07-29 14:39:52.952688+00:00incididunt lorem eiusmod do et ipsum et ut
- -

Warnings detected:

- -

Model: adipiscing magna lorem

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERipsum_et_incididunt_dout_sed_incididunt2025-07-29 14:39:52.952121+00:00aliqua tempor aliqua sit ipsum amet elit ipsum
- -

Model: sed amet adipiscing incididunt

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
ALIQUA_SIT_INCIDIDUNT_EIUSMODconsectetur_dolore_sit_sed_seddolore_ipsum2025-07-29 14:39:52.952259+00:00do aliqua ut incididunt consectetur consectetur sed et labore
- -

Model: ipsum ipsum amet tempor

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
OPC_LISTENNING_STOPPED__server_6sit_labore_sed_dolor_et_elitdolor_do_magna_et2025-07-29 14:39:52.952376+00:00magna consectetur do et dolor aliqua
- -

Model: sed ut

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERelit_et_ipsum_doloresed_ipsum2025-07-29 14:39:52.952426+00:00elit dolor dolore dolor eiusmod
- -

Model: eiusmod labore

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
OPC_CONNECTION_RETRY__server_3dolor_ut_dolor_sit_adipiscing_incididuntsit_tempor_dolore2025-07-29 14:39:52.952483+00:00ipsum consectetur magna elit dolore
- -

Model: do dolor tempor amet

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
REPORT_PARTITION_MANAGERsed_sit_do_tempor_uteiusmod_adipiscing2025-07-29 14:39:52.952637+00:00incididunt do do adipiscing dolore ut ut elit sit labore
- -

Infos detected:

- -

Model: sed consectetur ut

- - - - - - - - - - - - - - - - - - - - - -
Notification IDScheduleBlockTimestampMessage
TEMPOR_DO_DOLOR_TEMPOR_DOLORelit_tempor_dolore_consectetur_dolore_adipiscingmagna_labore_lorem2025-07-29 14:39:52.952199+00:00et et adipiscing lorem magna eiusmod labore do
- - - - - \ No newline at end of file diff --git a/input_sample.json b/input_sample.json deleted file mode 100644 index e8d3641..0000000 --- a/input_sample.json +++ /dev/null @@ -1,37 +0,0 @@ -[ -{ - "schedule_name": "orchestrator-test", - "pipelines_query": "SELECT pipelines.*, models FROM `pipelines` JOIN `models` ON KEYS pipelines.model_id;", - "opc_servers_query": "select META().id, opc_servers.* from `opc_servers`;" -}, -{ - "schedule_name": "orchestrator-test", - "pipelines_query": { - "collection": "pipelines", - "aggregation": [ - { - "$lookup": { - "from": "models", - "localField": "model_id", - "foreignField": "id", - "as": "model_docs" - } - }, - { - "$addFields": { - "models": { "$arrayElemAt": ["$model_docs", 0] } - } - }, - { - "$project": { - "model_docs": 0 - } - } - ] - }, - "opc_servers_query": { - "collection": "opc-servers", - "filters": {} - } -} -] \ No newline at end of file diff --git a/orchestrator/__init__.py b/orchestrator/__init__.py index e69de29..8b13789 100644 --- a/orchestrator/__init__.py +++ b/orchestrator/__init__.py @@ -0,0 +1 @@ + diff --git a/orchestrator/activities/activities.py b/orchestrator/activities/activities.py index ff36b17..56bf8bc 100644 --- a/orchestrator/activities/activities.py +++ b/orchestrator/activities/activities.py @@ -16,6 +16,23 @@ with workflow.unsafe.imports_passed_through(): class Activities( # Couchbase, TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres): + """ + Central activities orchestrator for Temporal workflow operations. + + This class combines multiple activity components including temporal management, + slot management, data formatting, MongoDB operations, email services, and + PostgreSQL operations. It provides a unified interface for all activity + operations required by the orchestration workflows. + + Args: + temporal_config (dict[str, Any]): Temporal server configuration + redis_config (dict[str, Any]): Redis server configuration + mongodb_config (dict[str, Any]): MongoDB connection configuration + email_config (dict[str, Any]): Email service configuration + postgres_config (dict[str, Any]): PostgreSQL database configuration + logger (Logger): Application logger instance + notification_handler (NotificationHandler): Notification management handler + """ def __init__(self, temporal_config: dict[str, Any], @@ -83,7 +100,11 @@ class Activities( # Couchbase, def shutdown(self): """ - Shutdown the MongoDB connection and clean up resources. + Shutdown all connections and clean up resources. + + This method gracefully shuts down all database connections, email + services, and other resources to ensure proper cleanup when the + application terminates. """ MongoDB.shutdown(self) Postgres.close(self) diff --git a/orchestrator/activities/couchbase.py b/orchestrator/activities/couchbase.py index 7a6e1cd..f7a8488 100644 --- a/orchestrator/activities/couchbase.py +++ b/orchestrator/activities/couchbase.py @@ -15,6 +15,25 @@ with workflow.unsafe.imports_passed_through(): class Couchbase(BaseActivity): + """ + Couchbase database operations activity (currently unused). + + This class provides Couchbase database connectivity and query operations + for Temporal workflows. It handles connection management, query execution, + and error reporting with automatic connection lifecycle management. + + Args: + connection_string (str): Couchbase cluster connection string + username (str): Couchbase authentication username + password (str): Couchbase authentication password + logger (Logger): Application logger instance + notification_handler (NotificationHandler): Notification management handler + + Note: + This class is currently commented out in the main Activities class + but maintained for potential future use. + """ + def __init__(self, connection_string: str, username: str, password: str, logger: Logger, notification_handler: NotificationHandler): diff --git a/orchestrator/activities/email.py b/orchestrator/activities/email.py index 8bd6718..783007c 100644 --- a/orchestrator/activities/email.py +++ b/orchestrator/activities/email.py @@ -19,6 +19,22 @@ with workflow.unsafe.imports_passed_through(): class Email(BaseActivity): + """ + Email service activity for sending workflow notifications. + + This class provides email sending capabilities including HTML email + generation, attachment handling, and SMTP connection management with + automatic reconnection for workflow notification delivery. + + Args: + sender_email (str): Email address for sending messages + sender_password (str): SMTP authentication password + smtp_server (str): SMTP server hostname + smtp_port (int): SMTP server port number + logger (Logger): Application logger instance + notification_handler (NotificationHandler): Notification management handler + """ + def __init__(self, sender_email: str, sender_password: str, smtp_server: str, smtp_port: int, logger: Logger, notification_handler: NotificationHandler): @@ -51,10 +67,24 @@ class Email(BaseActivity): @activity.defn(name="build_email_html") async def build_email_html(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Builds the email html for each receiver group. - input_data: - - receiver_groups (dict): The receiver groups. - - mail_type (str): The mail type. + Build HTML email content for configured receiver groups. + + This activity generates HTML email content for each receiver group + based on notification data and mail type. It processes notification + data through the email builder to create formatted HTML messages. + + Args: + input_data (dict[str, Any]): Activity input parameters. + Required fields: + - metadata (dict[str, Any]): Workflow execution metadata + - receiver_groups (dict[str, Any]): Receiver group configurations with notifications + - mail_type (str): Type of email (Alerts/Reports) + + Returns: + dict[str, Any]: Updated receiver groups with generated HTML content + + Raises: + Exception: If HTML generation fails """ metadata = input_data['metadata'] receiver_groups = input_data['receiver_groups'] @@ -147,10 +177,24 @@ class Email(BaseActivity): @activity.defn(name="send_email") async def send_email(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Sends an email to the receivers of each group. - input_data: - - receiver_groups (dict): The receiver groups. - - mail_type (str): The mail type. + Send email notifications to configured receiver groups. + + This activity sends HTML emails with attachments to all configured + receiver groups. It handles SMTP connection management, attachment + processing, and error reporting with automatic reconnection support. + + Args: + input_data (dict[str, Any]): Activity input parameters. + Required fields: + - metadata (dict[str, Any]): Workflow execution metadata + - receiver_groups (dict[str, Any]): Receiver groups with HTML content + - mail_type (str): Type of email being sent (Alerts/Reports) + + Returns: + dict[str, Any]: Updated receiver groups with sending status + + Raises: + Exception: If email sending fails for all groups """ metadata = input_data['metadata'] receiver_groups = input_data['receiver_groups'] diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index f3212ab..ba042b2 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -22,6 +22,21 @@ topic_separator = "\n ========== \n" class Formatters(BaseActivity): + """ + Schedule and slot configuration formatting 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. + + Args: + scouter_namespace (str): Scouter workflow namespace + laborious_namespace (str): Laborious workflow namespace + logger (Logger): Application logger instance + notification_handler (NotificationHandler): Notification management handler + """ + def __init__(self, scouter_namespace: str, laborious_namespace: str, diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index d456d7e..5f32540 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -41,6 +41,22 @@ def clear_mongo_id(docs: list) -> list: class MongoDB(BaseActivity): + """ + MongoDB operations activity for Temporal workflows. + + 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. + + Args: + connection_string (str): MongoDB connection string + database_name (str): Target database name + ttl_index_seconds (int): TTL index duration in seconds + logger (Logger): Application logger instance + notification_handler (NotificationHandler): Notification management handler + """ + def __init__(self, connection_string: str, database_name: str, ttl_index_seconds: int, logger: Logger, notification_handler: NotificationHandler): @@ -415,14 +431,25 @@ class MongoDB(BaseActivity): @activity.defn(name="load_latest_data") async def load_latest_data(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: """ - Loads the latest data from MongoDB. - input_data: - - metadata (dict): The metadata of the workflow. - - collection_name (str): The name of the collection to load data from. - - last_data_timestamp (str): The timestamp of the last data to load. - - base_data_filter (dict): The base data filter to apply to the query. - returns: - - data (list[dict]): The data loaded from MongoDB. + Load the latest data from MongoDB collection since a specified timestamp. + + This activity retrieves data from a MongoDB collection, optionally + filtering by timestamp to enable incremental data processing. It + handles connection management and provides comprehensive error reporting. + + Args: + input_data (dict[str, Any]): Activity input parameters. + Required fields: + - metadata (dict[str, Any]): Workflow execution metadata + - collection_name (str): Name of the MongoDB collection + - last_data_timestamp (str | None): Last processed timestamp for filtering + - base_data_filter (dict[str, Any]): Base query filter conditions + + Returns: + list[dict[str, Any]]: Retrieved data, or empty list if no data found + + Raises: + Exception: If MongoDB operation fails """ metadata = input_data['metadata'] collection_name = input_data['collection_name'] diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 862e258..aea49cc 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -15,6 +15,22 @@ with workflow.unsafe.imports_passed_through(): class SlotManager(Redis): + """ + Redis-based OPC slot management 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. + + Args: + host (str): Redis server hostname + port (int): Redis server port number + username (str): Redis authentication username + password (str): Redis authentication password + logger (Logger): Application logger instance + notification_handler (NotificationHandler): Notification management handler + """ def __init__(self, host: str, port: int, username: str, password: str, diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index 7e03a8b..95725a2 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -19,6 +19,22 @@ with workflow.unsafe.imports_passed_through(): class TemporalManager(BaseActivity): + """ + Temporal workflow and schedule management activity. + + This class manages Temporal schedules across multiple namespaces, + providing operations for schedule creation, updates, deletion, and + normalization. It handles connections to both scouter and laborious + namespaces for comprehensive workflow orchestration. + + Args: + host (str): Temporal server host address + scouter_namespace (str): Scouter workflow namespace + laborious_namespace (str): Laborious workflow namespace + logger (Logger): Application logger instance + notification_handler (NotificationHandler): Notification management handler + """ + def __init__(self, host: str, scouter_namespace: str, laborious_namespace: str, logger: Logger, notification_handler: NotificationHandler): diff --git a/orchestrator/metrics.py b/orchestrator/metrics.py index c7bc20e..2a6a56b 100644 --- a/orchestrator/metrics.py +++ b/orchestrator/metrics.py @@ -1,3 +1,11 @@ +""" +Prometheus metrics 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. +""" + from prometheus_client import Gauge, Counter APP_UP = Gauge( diff --git a/orchestrator/utils/converters.py b/orchestrator/utils/converters.py index 45bd436..eff3678 100644 --- a/orchestrator/utils/converters.py +++ b/orchestrator/utils/converters.py @@ -1,6 +1,23 @@ def parse_frequency(frequency: str) -> int: """ - Parse frequency string to seconds + Parse frequency string into seconds for Temporal schedule intervals. + + This function converts human-readable frequency strings into seconds + for use in Temporal schedule configurations. Supports seconds, minutes, + hours, and days notation. + + Args: + frequency (str): Frequency string with suffix: + - 's' for seconds (e.g., '30s') + - 'm' for minutes (e.g., '5m') + - 'h' for hours (e.g., '2h') + - 'd' for days (e.g., '1d') + + Returns: + int: Frequency converted to seconds + + Raises: + ValueError: If frequency format is invalid """ if frequency.endswith("s"): return int(frequency[:-1]) diff --git a/orchestrator/utils/email_builder.py b/orchestrator/utils/email_builder.py index c78be8b..1d74ece 100644 --- a/orchestrator/utils/email_builder.py +++ b/orchestrator/utils/email_builder.py @@ -6,6 +6,18 @@ import re class EmailBuilder: + """ + HTML email template builder for notification emails. + + This class handles the generation of HTML email content from notification + data using Jinja2 templates. It supports different email types (alerts, + reports) and notification levels (ERROR, WARNING, INFO) with customizable + templates and parameter replacement. + + Args: + logger (Logger): Application logger instance for error reporting + """ + def __init__(self, logger: Logger): self.logger = logger diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index b49a771..90bfdfb 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -211,7 +211,8 @@ def predictions_batch(config: dict[str, Any]): "config": {} } }, config.get('mlflow_predict_filters', [])), - "path_priority": path_priority + "path_priority": path_priority, + "predictions_storage_policy": config.get('predictions_storage_policy', 'lts:1') } diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index ea292b2..0b75851 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -9,16 +9,37 @@ with workflow.unsafe.imports_passed_through(): @workflow.defn(name="orchestrator") class Orchestrator: + """ + Main orchestrator workflow for pipeline and resource management. + + This workflow coordinates pipeline deployment and OPC server slot + management by retrieving configurations from MongoDB and Redis, + processing schedules, and deploying them to the Temporal server + and Redis infrastructure. + """ + @workflow.run async def run(self, input_data: dict[str, Any]): """ - Orchestrates the pipeline and slot management. Gets configuration from MongoDB and Redis, - creates the configuration and deploys the schedules and slots in the Temporal server and - Redis server. - input_data: - - schedule_name (str): The name of the schedule. - - pipelines_query (dict): The query to get the pipelines. - - opc_servers_query (dict): The query to get the OPC servers. + Execute the orchestration workflow for pipeline and slot management. + + This workflow retrieves pipeline configurations and OPC server data, + processes schedules and slot configurations, and deploys them to + the appropriate services. It handles creation, updates, and deletion + of schedules and slots based on current system state. + + Args: + input_data (dict[str, Any]): Workflow input parameters. + Required fields: + - schedule_name (str): Name of the orchestration schedule + - pipelines_query (dict[str, Any]): MongoDB query for pipeline configurations + - opc_servers_query (dict[str, Any]): MongoDB query for OPC server data + + Returns: + None: Workflow completes without return value + + Raises: + Exception: If orchestration operations fail """ input_data['workflow_name'] = 'orchestrator' diff --git a/orchestrator/workflows/reports.py b/orchestrator/workflows/reports.py index 321578c..e942296 100644 --- a/orchestrator/workflows/reports.py +++ b/orchestrator/workflows/reports.py @@ -9,20 +9,34 @@ with workflow.unsafe.imports_passed_through(): @workflow.defn(name="reports") class Reports: + """ + Reports workflow for sending scheduled notification summaries. + + This workflow processes and sends scheduled reports to configured + user groups. It loads notification data from MongoDB, filters it + by receiver group configurations, and sends formatted HTML reports + via email. + """ + @workflow.run async def run(self, input_data: dict[str, Any]): """ - Workflow to send reports to the users + Execute the reports workflow for scheduled notification delivery. + + This workflow loads all notifications from the notification queue, + applies receiver group filtering, and sends comprehensive HTML + reports to configured user groups. Args: - input_data (dict[str, Any]): Input data. It contains the following keys: - - schedule_name: str - Name of the schedule + input_data (dict[str, Any]): Workflow input parameters. + Required fields: + - schedule_name (str): Name of the report schedule Returns: - None + None: Workflow completes without return value Raises: - Exception: If the workflow fails + Exception: If report generation or delivery fails """ metadata = { 'metadata': { diff --git a/orchestrator/workflows/subworkflows/__init__.py b/orchestrator/workflows/subworkflows/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/orchestrator/workflows/subworkflows/load_notification_package.py b/orchestrator/workflows/subworkflows/load_notification_package.py index 5c83af7..9e2d35b 100644 --- a/orchestrator/workflows/subworkflows/load_notification_package.py +++ b/orchestrator/workflows/subworkflows/load_notification_package.py @@ -9,20 +9,39 @@ with workflow.unsafe.imports_passed_through(): @workflow.defn(name="load_notification_package") class LoadNotificationPackage: + """ + Subworkflow for loading notification data and configuration. + + This subworkflow retrieves notification packages from MongoDB and + loads receiver group configurations. It handles timestamp-based + filtering for incremental data processing and manages the data + required for notification workflows. + """ + @workflow.run async def run(self, input_data: dict[str, Any]): """ - Loads the notification package from the MongoDB collection "notification_queue" - and the sending configs from the MongoDB collection "receiver_groups". + Load notification package and sending configurations. - input_data: - - metadata (dict): The metadata of the workflow. + This subworkflow loads notifications from the MongoDB notification + queue using timestamp-based filtering and retrieves active receiver + group configurations. It updates the last processed timestamp in Redis. - returns: - - last_timestamp (str): The last timestamp of the notification package. - - notification_package (list[dict]): The notification package. - - sending_configs (list[dict]): The sending configs. - - mail_type (str): The mail type. + Args: + input_data (dict[str, Any]): Workflow input parameters. + Required fields: + - metadata (dict[str, Any]): Workflow execution metadata + - mail_type (str): Type of mail (Alerts/Reports) + - base_data_filter (dict[str, Any]): Base filter for notification query + + Returns: + dict[str, Any]: Package containing: + - last_timestamp (str | None): Last processed timestamp + - notification_package (list[dict]): Retrieved notifications + - sending_configs (list[dict]): Active receiver group configurations + + Raises: + Exception: If data loading fails """ metadata = input_data['metadata'] diff --git a/orchestrator/workflows/subworkflows/process_notifications.py b/orchestrator/workflows/subworkflows/process_notifications.py index e02fc12..121de9e 100644 --- a/orchestrator/workflows/subworkflows/process_notifications.py +++ b/orchestrator/workflows/subworkflows/process_notifications.py @@ -10,30 +10,37 @@ with workflow.unsafe.imports_passed_through(): @workflow.defn(name="process_notifications") class ProcessNotifications: + """ + Subworkflow for processing and sending notification emails. + + This subworkflow handles the email delivery process including HTML + generation, email sending, and logging to PostgreSQL. It processes + receiver groups and generates delivery reports for monitoring. + """ + @workflow.run async def run(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Processes the notifications. Builds the report html for each group and each model, - sends the report html to the receivers of each group, stores the sending log in the - postgres database "log_report", and returns the log report to the caller. + Process notifications and send emails to configured receiver groups. - input_data: - - metadata (dict): The metadata of the workflow. - - mail_type (str): The mail type. - - schema (str): The schema of the table. - - table_name (str): The name of the table. - - notification_package (list[dict]): The notification package. the format of each - notification package is: - { - 'group_name' (str) - 'group_members' (list[str]) - 'notifications' (dict) - { - 'model_name' (dict[str, list[dict]]) - } - } - returns: - - log_report (dict) + This subworkflow builds HTML email content, sends emails to all + receiver groups, and logs the delivery results to PostgreSQL for + monitoring and audit purposes. + + Args: + input_data (dict[str, Any]): Workflow input parameters. + Required fields: + - metadata (dict[str, Any]): Workflow execution metadata + - mail_type (str): Type of email being sent + - schema (str): PostgreSQL schema name for logging + - table_name (str): PostgreSQL table name for logging + - notification_package (dict[str, Any]): Receiver groups with notifications + + Returns: + dict[str, Any]: Log report of email delivery results + + Raises: + Exception: If notification processing fails """ metadata = input_data["metadata"] diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 09fae10..cd59478 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -197,7 +197,8 @@ def test_predictions_batch(mock_process_path_priority, } ], "path_priority": ["STOP", "CONTINUE", "REPEAT"], - "datetime_columns": ["timestamp"] + "datetime_columns": ["timestamp"], + "predictions_storage_policy": "erl:1" } result = predictions_batch(config) @@ -272,7 +273,8 @@ def test_predictions_batch(mock_process_path_priority, } }, "path_priority": ["STOP", "CONTINUE", "REPEAT"], - "datetime_columns": ["timestamp"] + "datetime_columns": ["timestamp"], + "predictions_storage_policy": "erl:1" } assert result == expected