# SIENTIA DataOps Orchestrator Temporal

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.

## Features

### 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

### 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 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.

### Architecture Principles

#### 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

#### 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

#### 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 (`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**:
```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}
  }
}
```

#### Architecture

```mermaid
flowchart LR
    subgraph "Parallel Data Loading"
        A[1. aggregate_documents_in_mongodb]
        B[2. find_documents_in_mongodb<br/>OPC Servers]
        C[3. find_documents_in_mongodb<br/>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**:
```json
{
  "schedule_name": "error_alerts",
  "notification_ttl": 3600,
  "sent_ttl": 7200
}
```

#### Architecture

```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**:
```json
{
  "schedule_name": "reports",
}
```

#### 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 (`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**:
```json
{
  "metadata": {"workflow_name": "alerts", "schedule_name": "error_alerts"},
  "mail_type": "Alerts",
  "base_data_filter": {"level": "ERROR"}
}
```

**Returns**:
- `last_timestamp` (str | None): Last processed timestamp
- `notification_package` (list[dict]): Retrieved notifications
- `sending_configs` (list[dict]): Active receiver group configurations

**Architecture**:

```mermaid
flowchart LR
    subgraph "Parallel Loading"
        A[1. get_last_data_timestamp]
        B[2. find_documents_in_mongodb<br/>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**:
```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` (list[dict]): Detailed delivery status for each notification

**Architecture**:

```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)]

```

### 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 <repository-url>
   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
```

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.