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:

{
  "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

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:

{
  "schedule_name": "error_alerts",
  "notification_ttl": 3600,
  "sent_ttl": 7200
}

Architecture

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": "reports",
}

Architecture

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:

{
  "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:

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:

{
  "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:

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

   git clone <repository-url>
   cd sientia-dataops-orchestrator_temporal
   

2. Create virtual environment

   python3.11 -m venv venv
   source ./venv/bin/activate
   

3. Install dependencies

   pip install -r requirements.txt
   

4. Configure environment variables

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

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

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

# 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

{
    "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

{
    "schedule_name": "alerts",
    "notification_ttl": 5*60,
    "sent_ttl": 10*60
}

Reports Workflow

{
    "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

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

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.