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.

📑 Table of Contents

• Features
  • Core Functionality
  • Advanced Capabilities
• Architecture
  • Architecture Principles
  • Task Queue Isolation
• Workflows
  • Orchestrator Workflow
  • Alerts Workflow
  • Reports Workflow
  • Subworkflows
    • Load Notification Package
    • Process Notifications
• Notification Filtering System
• Prerequisites
• Installation
• How to Run
• Configuration
• Monitoring and Metrics
• Testing
• Code Quality & Validation
• Development
• Troubleshooting
• Performance Tuning
• Contributing
• License
• Support

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
• Intelligent Notification Filtering: Advanced filtering system with:
  • User group-based notification filtering with custom policies
  • TTL-based duplicate prevention for alerts
  • Ignore lists for specific notifications
  • Persistent alert detection for ongoing issues
• Auto-scaling Workers: Multiple worker instances with task queue isolation
• Comprehensive Logging: Structured logging with PostgreSQL audit trails
• Prometheus Metrics: Real-time monitoring and alerting integration
• Template-based Email Generation: Jinja2-powered HTML email templates

Architecture

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, Prometheus metrics, and application lifecycle
• Workflow Layer: Orchestrates business logic and process coordination with subworkflows for modularity
• Activity Layer: Implements specific operations and external system interactions with error handling
• Data Layer: Handles data persistence (MongoDB, PostgreSQL), caching (Redis), and external service connections

2. Task Queue Isolation

• Orchestrator Queue: Pipeline and resource management workflows (orchestrator workflow)
• Alerts Queue: Real-time error notification workflows (alerts workflow, load_notification_package, process_notifications subworkflows)
• Reports Queue: Scheduled reporting and summary workflows (reports workflow, load_notification_package, process_notifications subworkflows)

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. Parallel Data Loading: Concurrently loads pipelines, orchestrated schedules, OPC slots, active ingestors, and the tag database (connectors and connector_tags, the only source of the OPC servers and tags of the slot and the schedules)
2. Parallel Processing: Formats orchestrated schedules and processes new schedules and slots
3. Parallel Config Creation: Creates schedule and slot action configurations, normalizes schedules, and creates collections with TTL indexes
4. Parallel Operations: Executes slot deletions, slot updates, schedule deletions, schedule creations, and schedule updates concurrently
5. Parallel Reports & Timestamps: Generates orchestration reports and updates MongoDB timestamps for created/updated/deleted pipelines

Input Parameters (minimal: each pipeline must already carry model.name; see "Workflow Configuration" for the $lookup on models that adds it):

{
  "schedule_name": "hourly_orchestration",
  "pipelines_query": {
    "collection": "pipelines",
    "aggregation": [
      {"$match": {"active": true}},
      {"$sort": {"updated_at": -1}}
    ]
  }
}

Optional: connectors_query and connector_tags_query override the tag database aggregations. The defaults live in the code (orchestrator/workflows/orchestrator.py), so the schedule input does not need them:

{
  "connectors_query": {
    "collection": "connectors",
    "aggregation": [
      {"$match": {"type": "opc_ua", "status": "Active"}},
      {"$lookup": {"from": "opc_certificates",
                   "let": {"certificate_id": "$security.client_certificate_id"},
                   "pipeline": [
                     {"$match": {"$expr": {"$eq": ["$certificate_id", "$$certificate_id"]},
                                 "status": "active", "kind": "client"}},
                     {"$project": {"_id": 0, "certificate_id": 1, "application_uri": 1,
                                   "cert.bucket": 1, "cert.object_name": 1,
                                   "private_key.bucket": 1, "private_key.object_name": 1}}
                   ],
                   "as": "certificates"}},
      {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "subscription_period_ms": 1,
                    "security.mode": 1, "security.policy": 1,
                    "security.client_certificate_id": 1, "certificates": 1}}
    ]
  },
  "connector_tags_query": {
    "collection": "connector_tags",
    "aggregation": [
      {"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1,
                    "category": 1, "data_type": 1, "status": 1}}
    ]
  }
}

The workflow puts {"$match": {"tag_id": {"$in": [...]}}} in front of connector_tags_query, so it loads only the tags the scouter read tags and predictions_batch write tags reference. When no pipeline references a tag_id, the tag database is not queried. An override should keep a $project of these fields: any other attribute may hold a BSON type a Temporal payload cannot carry (such as ObjectId).

The referenced tags are loaded whatever their category and status: the code uses only the active ones (Read for the slot and the scouter names, Write / Prediction Confidence for the write tags), and a change to a referenced tag that reaches a schedule redeploys it through its config_hash (see "Redeploy on changes" below). A document without status counts as active.

OPC slots (slot v2, contract C1)

process_slots writes one slot per active ingestor to Redis (slot:opc_tags:{n}). The OPC Gateway reads it (SIENTIAPDE-1974). Each server carries its connection fields, its security and the MinIO objects of its client certificate, and each tag address carries the list of names it is written under:

{
  "kepserver": {
    "server_id": "1", "name": "kepserver", "url": "opc.tcp://host:49320",
    "subscription_period_ms": 7500.0,
    "security": {"mode": "SignAndEncrypt", "policy": "Basic256Sha256"},
    "certificate": {
      "certificate_id": 1, "application_uri": "urn:sientia:opc-ingestor",
      "cert": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/opc-ingestor-cert.pem"},
      "private_key": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/opc-ingestor-key.pem"}
    },
    "tags": {
      "ns=2;s=Channel1.Device1.Tag1": {
        "frequency": 15000,
        "active_read": {"enabled": false},
        "names": [
          {"collection": "raw_model_a", "tag_name": "Counter",  "model_id": "1", "model_name": "model a"},
          {"collection": "raw_model_b", "tag_name": "Contador", "model_id": "2", "model_name": "model b"}
        ]
      }
    }
  }
}

Each scouter read_tags entry references its tag by tag_id ({"tag_id": 7, "frequency": 15000}). connectors is the only source of the OPC servers:

Field	Source
Address	connector_tags.address
Server (server_id, name, url, subscription_period_ms, security.mode, security.policy)	The tag's connector_tags.connector_id in connectors (server_id = connector_id)
certificate	The opc_certificates document of the connector's security.client_certificate_id, when it is active and of kind client (connectors_query's $lookup)
Name	The read tag's tag_name, else the connector_tags one

Rules:

• One entry per address. An address read by N pipelines is read once and written to N collections: one names item per pipeline (collection = raw_{schedule_name}).
• Same name source as the scouter. resolve_tag_names names both the slot's names and the scouter's model_tags, so the gateway always writes the name the scouter looks for.
• frequency: the lowest among the pipelines of the address, as an integer. Every read tag needs one.
• tag_id: required, an integer like in connector_tags. A read tag whose tag_id is missing, null or not an integer ("7" included) is left out and notified as invalid.
• subscription_period_ms: the connector's value. A connector without one gets half its lowest tag frequency in the slot.
• security: the connector's mode and policy, the source of the OPC security (use_security and server_uri no longer exist). A connector without them is left out like an incomplete one.
• certificate: certificate_id, application_uri (the SAN URI of the certificate, used as the client application URI) and the bucket/object_name of cert and private_key; the gateway downloads them from MinIO. The certificate dates, sha256 and audit fields stay out of the slot. There is no default certificate: null means the gateway connects without security, and a server that requires a certificate refuses the connection. A connector whose client_certificate_id is set but has no active client certificate (missing id, retired, another kind or an incomplete document) still goes into the slot with certificate: null, and is reported (see below).
• active_read: off by default. It is on when any pipeline sets read_tags[].active_read.enabled, with the lowest cycle_multiple.
• Server keys: two different connectors with the same name get distinct slot keys (name, then name:server_id).
• Slot division: the addresses are split across the slots, sorted by server and address, and the names of an address always stay in the same slot.
• Unresolved tags: a read tag that cannot be resolved is left out of the slot, and the rest of the slot is still written. It is notified once per run:
  • ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION: the tag_id is not an active Read tag, or its document lacks connector_id, address or tag_name;
  • ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION: the connector is missing or inactive, or lacks name, url, security.mode or security.policy;
  • ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION: the read tag has no integer tag_id, no integer frequency, or an enabled active_read without an integer cycle_multiple; or its pipeline lacks schedule_name, model_id or model.name (the value is then the schedule_name).
  • ORCHESTRATOR_CERTIFICATE_NOT_FOUND_DURING_SLOT_CONFIGURATION (WARNING): the connector's client_certificate_id has no active client certificate, so its tags go into the slot without a certificate. Most likely a registration error.
• Only changed slots are written. The slot is built in a stable order, so unchanged pipelines produce an identical slot, and the gateway does not re-subscribe.

Write tags (predictions_batch, through the tag database)

A predictions_batch (or xgboost_predictions_batch) references its write tags by tag_id only, exactly like the read tags. Everything else comes from the tag database, so the pipeline does not say which server, address or type a tag has:

{"workflow_type": "predictions_batch", "write_tags": [{"tag_id": 101}, {"tag_id": 102}]}

process_schedules turns them into the opc_output_config the Laborious already reads:

{"plant": {
  "prediction_tags": {"ns=3;i=2001": {"data_type": "double"}},
  "confidence_tags": {"ns=3;i=2002": {"data_type": "float"}}}}

Field	Source
Server key	connector_tags.connector_id. The Laborious publishes it as server_id on NATS, and the OPC Gateway resolves it in connectors
prediction_tags / confidence_tags	connector_tags.category: Write / Prediction Confidence
Address	connector_tags.address
data_type	connector_tags.data_type (float, double, real, int, integer, bool, boolean, str, string), else float

A write tag that cannot be resolved is left out, the rest of the pipeline is still scheduled, and it is notified once per run (values are schedule_name/tag_id):

• ORCHESTRATOR_WRITE_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION: the tag_id is not an active Write / Prediction Confidence tag, or its document lacks connector_id, address, tag_name or category;
• ORCHESTRATOR_WRITE_TAG_CONNECTOR_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION: the tag's connector is missing, inactive or incomplete in connectors;
• ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION: the entry has no integer tag_id, or the tag's data_type is not one the gateway accepts.

Redeploy on changes (updated_at and config_hash)

orchestrated_schedules keeps, per schedule, when it was last deployed (updated_at) and the config_hash it was deployed with: a SHA-256 of the schedule as the Orchestrator builds it (the workflow input, frequency and offset, with tags and connectors already resolved), without updated_at. A schedule is redeployed when either:

• the pipeline's updated_at is newer than its last deploy. Bumping updated_at by hand forces a redeploy even when nothing else changed; or
• its config_hash differs from the stored one. Any change that reaches the schedule redeploys it, whether or not someone bumped an updated_at: a pipeline edited by hand, a tag renamed, moved, retyped, deactivated or deleted in connector_tags, a connector changed, or a new Orchestrator version that builds the schedule differently.

A change that does not reach the schedule (e.g. a tag's description) does not redeploy. The slot is compared by content and follows on its own. A pipeline without updated_at is still redeployed on every run. Schedules deployed before config_hash existed have none stored, so the first run of this version redeploys all of them once and records it.

Deploying

1. Run the migrations that create the tag database and add data_type to the connector_tags validator (sientia-dataops-database-migrations).
2. Every scouter read_tags and every predictions_batch / xgboost_predictions_batch write_tags entry is a tag_id of connector_tags, as an integer. Anything else is left out and notified.
3. Deploy this version. Its first run redeploys every schedule once, to record their config_hash (see above).

Architecture

flowchart LR
    subgraph "Parallel Data Loading"
        A[1. aggregate_documents_in_mongodb]
        B[2. aggregate_documents_in_mongodb<br/>connectors + connector_tags]
        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: Subworkflow loads ERROR-level notifications from MongoDB with timestamp filtering
2. Configuration Loading: Loads active receiver group configurations from MongoDB
3. Alert Filtering: Applies TTL-based filtering to identify core_alerts (new) and persistent_alerts (ongoing issues)
4. Group Filtering: Filters notifications by user group content policies and ignore lists
5. Email Processing: Subworkflow generates HTML emails and sends to receiver groups
6. Cache Storage: Stores sent notification cache in Redis with configurable TTL to prevent duplicates
7. Audit Logging: Logs delivery results to PostgreSQL for monitoring

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. Notification Loading: Subworkflow loads all notifications from MongoDB (any level) with timestamp filtering
2. Configuration Loading: Loads active receiver group configurations from MongoDB
3. Report Filtering: Filters notifications by user group content policies (must include "reports") and ignore lists
4. Email Processing: Subworkflow generates HTML reports organized by notification level and model, then sends to receiver groups
5. Audit Logging: Logs delivery results to PostgreSQL for monitoring and tracking

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

Key Activities:

• get_last_data_timestamp: Retrieves last processed timestamp from Redis for incremental processing
• find_documents_in_mongodb: Loads active receiver group configurations from MongoDB
• load_latest_data: Loads notifications with timestamp filtering from notification_queue collection
• put_last_data_timestamp: Updates last processed timestamp in Redis with 5-hour TTL

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 using Jinja2 templates
• Email Delivery: Sends emails with attachment support and error handling
• Audit Logging: Records delivery status and metrics in PostgreSQL
• Error Recovery: Handles SMTP failures with detailed error reporting
• Template Support: Uses customizable HTML templates for different email types

Input Parameters:

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

Key Activities:

• build_email_html: Generates HTML content using Jinja2 templates organized by notification level and model
• send_email: Delivers emails to receiver groups with attachment support and automatic SMTP reconnection
• format_log_report: Formats delivery results into DataFrame structure for database storage, aggregating by notification ID and trigger
• export_data_to_postgres: Stores audit logs in PostgreSQL with timestamp conversion

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

🔔 Notification Filtering System

The orchestrator includes an advanced notification filtering system that prevents alert spam and ensures relevant notifications reach the appropriate user groups.

Alert Filtering (filter_notification_alerts)

• Purpose: Filters error-level notifications for immediate alerts
• TTL Management: Prevents duplicate alerts using configurable time-to-live settings
• Persistent Detection: Identifies ongoing issues that require escalation
• Group-based Filtering: Routes notifications to appropriate receiver groups
• Ignore Lists: Supports notification exclusion per group

Report Filtering (filter_notification_reports)

• Purpose: Filters all notification levels for comprehensive scheduled reports
• Comprehensive Coverage: Includes ERROR, WARNING, INFO, and DEBUG levels
• Group Customization: Applies different content policies per receiver group with ignore list support
• Scheduled Processing: Designed for regular report generation without TTL-based duplicate prevention
• Duplicate Prevention: Prevents duplicate notifications within the same report using trigger and notification ID keys

Notification Caching (store_notification_cache)

• Purpose: Manages Redis-based notification cache for TTL enforcement
• TTL Support: Configurable expiration times for different notification types
• Duplicate Prevention: Ensures notifications aren't sent repeatedly within TTL window
• Key Management: Uses structured keys for efficient cache lookups

Key Components

Worker (orchestrator/worker/worker.py)

• Purpose: Main application orchestrator managing Temporal workers and task queues
• Responsibilities:
  • Temporal client initialization and connection management with SDK metrics
  • Worker lifecycle management and graceful shutdown via sientia_do.temporal.worker.prepare_worker
  • Task queue configuration (orchestrator, alerts, reports) with dedicated workers on <workflow>-<runtime>-queue
  • Prometheus metrics server initialization on HTTP_METRICS_PORT
  • Temporal SDK metrics server initialization on HTTP_SDK_METRICS_PORT
  • Notification handler setup and configuration
• Key Features:
  • Multi-queue worker management with three dedicated workers (orchestrator, alerts, reports)
  • RUNTIME env var (default legacy) passed to every prepare_worker call
  • Sync blocking activities run on prepare_worker's activity_executor thread pool; workflows stay async def
  • Application health metrics (app_up gauge) for Kubernetes liveness/readiness probes
  • Graceful shutdown with cleanup procedures for all connections and exit code propagation to Kubernetes
  • Parallel worker execution using asyncio.gather

Activities (orchestrator/activities/)

• Activities: Main activity orchestrator combining all operations (TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres)
• TemporalManager: Temporal schedule CRUD operations across scouter and laborious namespaces with search attributes
• SlotManager: Redis-based OPC slot and cache management with notification filtering and TTL-based duplicate prevention
• MongoDB: Document operations, aggregations, timestamp management, and TTL index creation
• Email: SMTP operations with HTML generation, attachment support, and automatic reconnection handling
• Formatters: Configuration processing, slot distribution algorithms, notification filtering for reports, and schedule/slot orchestration reporting
• Postgres: PostgreSQL operations for audit logging and data export (via sientia-dataops-library)

Utilities (orchestrator/utils/)

• Connectors Configuration: Database and service configuration management from environment variables
• Email Builder: HTML email template generation and formatting using Jinja2 templates with support for alerts and reports
• Orchestrator Functions: Pipeline configuration transformation utilities supporting:
  • scouter: OPC UA data collection workflows
  • pi_web_api_scouter: PI Web API data collection workflows
  • predictions_batch: ML prediction workflows with OPC write-back and multi-stage filtering
  • minimal_retrain: Model retraining workflows with SQL queries
  • drift: Data drift detection workflows
  • simple_metrics: Model performance metrics computation workflows
• Converters: Data type conversion and validation utilities including frequency parsing for Temporal schedules
• Templates: HTML email templates for alerts and reports (email_template.html, general_template.html)

📋 Prerequisites

System Requirements

• 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
RUNTIME	Runtime slice for orchestrator worker task queues (orchestrator-<runtime>-queue, etc.)	legacy	No
ACTIVITY_EXECUTOR_MAX_WORKERS	Thread pool size for sync activities (all workers)	200	No
ORCHESTRATOR_ACTIVITY_EXECUTOR_MAX_WORKERS	Per-worker override for the Orchestrator worker	falls back to ACTIVITY_EXECUTOR_MAX_WORKERS	No
ALERTS_ACTIVITY_EXECUTOR_MAX_WORKERS	Per-worker override for the Alerts worker	falls back to ACTIVITY_EXECUTOR_MAX_WORKERS	No
REPORTS_ACTIVITY_EXECUTOR_MAX_WORKERS	Per-worker override for the Reports worker	falls back to ACTIVITY_EXECUTOR_MAX_WORKERS	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
POSTGRES_MIN_CONNECTIONS	PostgreSQL minimum connections	10	No
POSTGRES_MAX_CONNECTIONS	PostgreSQL maximum connections	40	No
MONGODB_TTL_INDEX_HOURS	MongoDB TTL index expiration in hours	1	No
PROJECT_NAME	Project name for metrics and logging	sientia-orchestrator	No
LOG_LEVEL	Application logging level	INFO	No
KAFKA_BOOTSTRAP_SERVERS	Kafka bootstrap servers	-	No

MongoDB Collections

Collection	Read / written by	Purpose
pipelines	Orchestrator (pipelines_query)	One document per Temporal schedule (schedule_name, workflow_type, model_id, active, frequency, ...)
models	Orchestrator ($lookup in pipelines_query)	Gives each pipeline its model (name, active, model_config)
connectors	Orchestrator (connectors_query)	Tag database: the OPC UA servers (connector_id, type, name, url, security, subscription_period_ms, status)
connector_tags	Orchestrator (connector_tags_query)	Tag database: the tags (tag_id, connector_id, address, tag_name, category, data_type, status)
opc_certificates	Orchestrator ($lookup in connectors_query)	The client certificate of each connector (security.client_certificate_id → certificate_id), only when active and of kind client: application_uri and the MinIO objects of cert and private_key go into the slot
orchestrated_schedules	Orchestrator (read and written)	Per deployed schedule: schedule_name, namespace, updated_at (last deploy) and config_hash
raw_{schedule_name}	Orchestrator (created)	One per scouter topic, where the OPC Gateway writes its tags. Created with a TTL index on inserted_at (MONGODB_TTL_INDEX_HOURS)
notification_queue	Alerts / Reports (read)	Notifications (notification_id, trigger, level, model_id, timestamp, ...), loaded incrementally by timestamp
receiver_groups	Alerts / Reports (read, active: true)	Who receives what (see below)

Tags in a pipeline

Pipelines reference tags only by tag_id (an integer, as in connector_tags). Servers, addresses and data types come from the tag database, never from the pipeline.

Scouter (scouter namespace), workflow_type: "scouter":

{
  "schedule_name": "scouter-linreg",
  "workflow_type": "scouter",
  "model_id": "1",
  "active": true,
  "frequency": "1m",
  "read_tags": [
    {
      "tag_id": 7,
      "frequency": 15000,
      "tag_name": "Pressure",
      "aggr_func": "avg",
      "data_range": [0, 100],
      "active_read": {"enabled": true, "cycle_multiple": 2}
    }
  ]
}

Field	Required	Meaning
tag_id	Yes	An active Read tag of connector_tags
frequency	Yes	Read period in ms, an integer. The slot uses the lowest among the pipelines of the address
tag_name	No	Overrides the connector_tags name: the name the gateway writes and the scouter looks for
aggr_func	No	Default lts
data_range	No	Default [-100, 100]
active_read	No	Off by default. When enabled, cycle_multiple must be an integer (default 1)

Laborious (laborious namespace): only predictions_batch and xgboost_predictions_batch reference tags; minimal_retrain, drift and simple_metrics have none.

{
  "schedule_name": "pred-linreg",
  "workflow_type": "predictions_batch",
  "model_id": "1",
  "active": true,
  "frequency": "1m",
  "query": "SELECT ...",
  "write_tags": [{"tag_id": 101}, {"tag_id": 102}]
}

Each write_tags entry is only a tag_id, of an active Write (prediction) or Prediction Confidence tag. The Orchestrator builds opc_output_config from them (see "Write tags" above) and overwrites any opc_output_config stored in the pipeline.

The connector_tags documents those pipelines reference:

{"tag_id": 7,   "connector_id": "plant", "address": "ns=3;i=1001", "tag_name": "Pressure", "category": "Read", "status": "Active"}
{"tag_id": 101, "connector_id": "plant", "address": "ns=3;i=2001", "tag_name": "PRED-101", "category": "Write", "data_type": "double", "status": "Active"}
{"tag_id": 102, "connector_id": "plant", "address": "ns=3;i=2002", "tag_name": "CONF-101", "category": "Prediction Confidence", "status": "Active"}

pi_web_api_scouter (scouter namespace) does not use the tag database: its read_tags is an object keyed by tag name, and it is not part of the OPC slot:

{
  "workflow_type": "pi_web_api_scouter",
  "read_tags": {"Pressure": {"webid": "F1DP...", "aggr_func": "lts", "data_range": [0, 100]}},
  "pi_web_api_config": {"endpoint": "https://pi/piwebapi", "period": "*-1d", "max_count": 1, "api_timeout": 30}
}

Receiver groups

{
  "group_name": "admin_team",
  "active": true,
  "members": ["admin@example.com"],
  "contents": ["core_alerts", "persistent_alerts", "reports"],
  "levels": ["ERROR", "WARNING"],
  "model_ids": ["1"],
  "ignore_models_id": ["2"],
  "ignore": ["SOME_NOTIFICATION_ID"]
}

Field	Meaning
contents	What the group gets: core_alerts / persistent_alerts (Alerts workflow), reports (Reports workflow)
levels	Optional allow-list of notification levels. Absent = every level the workflow loads
model_ids	Optional allow-list of model_ids
ignore_models_id	Optional deny-list of model_ids (applies together with model_ids)
ignore	Optional deny-list of notification_ids

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": {
                    "model": {
                        "$arrayElemAt": [
                            "$model_docs",
                            0
                        ]
                    }
                }
            },
            {
                "$match": {
                    "model.active": True
                }
            },
            {
                "$project": {
                    "model_docs": 0
                }
            }
        ]
    },
    "connectors_query": {
        "collection": "connectors",
        "aggregation": [
            {"$match": {"type": "opc_ua", "status": "Active"}},
            {"$lookup": {"from": "opc_certificates",
                         "let": {"certificate_id": "$security.client_certificate_id"},
                         "pipeline": [
                           {"$match": {"$expr": {"$eq": ["$certificate_id", "$$certificate_id"]},
                                       "status": "active", "kind": "client"}},
                           {"$project": {"_id": 0, "certificate_id": 1, "application_uri": 1,
                                         "cert.bucket": 1, "cert.object_name": 1,
                                         "private_key.bucket": 1, "private_key.object_name": 1}}
                         ],
                         "as": "certificates"}},
            {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "subscription_period_ms": 1,
                          "security.mode": 1, "security.policy": 1,
                          "security.client_certificate_id": 1, "certificates": 1}}
        ]
    },
    "connector_tags_query": {
        "collection": "connector_tags",
        "aggregation": [
            {"$project": {"_id": 0, "tag_id": 1, "connector_id": 1, "address": 1, "tag_name": 1,
                          "category": 1, "data_type": 1, "status": 1}}
        ]
    }
}

connectors_query and connector_tags_query are optional: the values above are the defaults in the code (see "OPC slots" and "Write tags" above).

The $lookup must end in a model field (singular): every pipeline type reads model.name (and model.model_config, when present), so a pipeline without model fails to build its schedule.

Simple metrics: parallel windows & adaptive frequency

One pipelines doc per window, all with the same model_id. Name them <model>-simple_metrics-<interval_minutes>m; each doc becomes its own Temporal schedule. Thresholds are shared through model.model_config. The adaptive state is per doc.

{
    "schedule_name": "model-x-simple_metrics-60m",
    "workflow_type": "simple_metrics",
    "model_id": "model-x",
    "active": true,
    "interval_minutes": 60,
    "frequency": "1h",
    "frequency_alert": "5m"
}

Field	Meaning	Default
frequency_alert	Frequency while escalated, same format as frequency. Absent = adaptive mode off. Below 1m (or unparseable), the orchestrator sends a WARNING each time it creates or updates the schedule. A value below 1m is kept; an unparseable value is ignored (the schedule keeps frequency)	—
adaptive_state	{escalated, healthy_streak}, written only by laborious (with updated_at as a BSON datetime when escalated flips)	{false, 0}
model.model_config.simple_metrics_thresholds	{<metric>_max, <metric>_min} shared by all windows	—
model.model_config.simple_metrics_min_samples	Windows with fewer rows neither alert nor escalate, and count toward recovery	—

Escalation, recovery and the per-model alert are implemented in laborious (SIENTIAPDE-2189, SIENTIAPDE-2191); the orchestrator only picks frequency_alert while adaptive_state.escalated is true.

Rules:

• Escalation: laborious escalates on the first breach.
• Recovery: after min(12, ceil(interval_minutes / frequency_alert)) consecutive runs without one. Empty windows count as runs without a breach.
• Alerts are per model (trigger <model_id>-simple_metrics), so several windows seeing the same breach produce one alert.
• Delay: the schedule switches on the orchestrator tick after laborious flips the state, at most one tick later.

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 gauge (1=healthy, 0=unhealthy) labeled by pod_id
• email_sent_count: Email delivery counter labeled by pod_id, model_name, pipeline_name, and email_group

Workflow Metrics

• Schedule creation, update, and deletion success rates (via notification reports)
• Notification processing times and error rates (via PostgreSQL audit logs)
• Resource allocation and slot management metrics (via notification reports)
• Temporal SDK metrics exposed on HTTP_SDK_METRICS_PORT (default: 9091)

Database Metrics

• MongoDB query performance and connection health (via sientia-dataops-library)
• Redis operation counts and response times (via sientia-dataops-library)
• PostgreSQL export operations and audit log metrics (via sientia-dataops-library)

🧪 Testing

Test Structure

tests/
├── orchestrator/           # Orchestrator workflow tests
│   ├── test_activities.py
│   ├── test_email.py
│   ├── test_formatters.py
│   ├── test_mongo_db.py
│   ├── test_slot_manager.py
│   ├── test_temporal_manager.py
│   └── test_workflows.py
├── activities/            # Activity implementation tests
├── utils/                 # Utility function tests
└── integration/           # End-to-end workflow tests

Test Coverage

The project maintains comprehensive test coverage including:

• Activity Tests: Unit tests for all activity classes
• Workflow Tests: Integration tests for workflow orchestration
• Utility Tests: Tests for configuration builders and converters
• Database Tests: Tests for MongoDB, Redis, and PostgreSQL operations
• Email Tests: Tests for email generation and delivery
• Notification Tests: Tests for filtering and caching logic

Test Execution

# Install test dependencies
pip install -r requirements-dev.txt

# Run unit tests with coverage (default testpaths=tests; E2E excluded)
pytest --cov=orchestrator --cov-report=html

# Run specific test modules
pytest tests/orchestrator/activities/test_mongo_db.py
pytest tests/orchestrator/workflows/test_orchestrator.py

End-to-end tests

E2E tests live in e2e/ and require Docker (testcontainers). They are not collected by default pytest at the repo root.

source ./venv/bin/activate
pip install -r requirements-dev.txt
pytest e2e/ --override-ini testpaths=e2e -m e2e -v

Coverage is kept separate from unit tests: set COVERAGE_FILE=.coverage.e2e when measuring E2E coverage (see e2e/README.md).

🛡️ Code Quality & Validation

Overview

Since Python is not compiled, this project ships a validation workflow to catch issues early. Use the validate.sh script to run formatting, linting, type checks, security analysis, and tests in one command.

Validation Tools

• Ruff: formatting and linting (fast, replaces Black/Flake8)
• mypy: static typing checks
• Bandit: security static analysis
• pytest: unit/integration tests with coverage

Tools Installation

pip install -r requirements-dev.txt

Complete Validation (recommended)

./validate.sh

What validate.sh does:

1. Checks formatting with Ruff
2. Lints code with Ruff
3. Runs mypy type checking
4. Runs Bandit security analysis
5. Executes pytest with coverage (generates HTML report)

Exit codes are propagated so CI can fail fast when quality gates are not met.

Individual Commands

# 1) Format check
ruff format --check orchestrator/ tests/

# 2) Lint
ruff check orchestrator/ tests/

# 3) Type check
mypy orchestrator/

# 4) Security
bandit -r orchestrator/ -ll

# 5) Tests with coverage
pytest tests/ --cov=orchestrator --cov-report=html

Automatic Fixes

# Apply formatting
ruff format orchestrator/ tests/

# Autofix common lint issues
ruff check --fix orchestrator/ tests/

Configuration

Tooling is configured in pyproject.toml (lint rules, formatting, typing). Adjust thresholds and rules there as needed.

🔧 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 and notification filtering
│   ├── mongo_db.py         # MongoDB operations
│   ├── email.py            # Email service operations
│   ├── formatters.py       # Configuration formatting and report filtering
├── workflows/               # Temporal workflow definitions
│   ├── orchestrator.py     # Main orchestration workflow
│   ├── alerts.py           # Error alert workflow
│   ├── reports.py          # Scheduled report workflow
│   └── subworkflows/       # Sub-workflow implementations
│       ├── load_notification_package.py  # Notification data loading
│       └── process_notifications.py      # Email processing and delivery
├── worker/                  # Worker implementation
│   └── worker.py           # Main worker orchestrator
├── utils/                   # Utility functions
│   ├── connectors_config.py # Database configuration
│   ├── email_builder.py    # Email template generation
│   ├── pipeline_configs.py # Schedule input builders per pipeline type
│   ├── tag_database.py     # Tag database indexes, OPC read names and write tags
│   ├── opc_slots.py        # OPC slot v2 (contract C1)
│   ├── converters.py       # Data type conversion utilities
│   └── templates/          # Email HTML templates
│       ├── email_template.html    # Report email template
│       └── general_template.html  # General email template
└── metrics.py               # Prometheus metrics definitions

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

5. E2E / testcontainers leftovers

   • If a run is interrupted, containers may keep running. List and remove them:

   docker ps -a --filter label=org.testcontainers=true
   docker rm -f $(docker ps -aq --filter label=org.testcontainers=true)
   

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

---

Runtime queues and migration

• Queue naming: Downstream pipeline schedules target <workflow_type>-<runtime>-queue (for example scouter-legacy-queue). The orchestrator's own workers register on orchestrator-<runtime>-queue, alerts-<runtime>-queue, and reports-<runtime>-queue.
• Pipeline runtime: Each pipeline JSON may include runtime (default legacy via common_config). Schedules without runtime use legacy.
• Schedule updates: Changing a pipeline's runtime does not update an existing schedule's task_queue in place; the next orchestrator tick deletes and recreates the schedule on the new queue.
• Rollout order: Deploy scouter/laborious worker fleets bound to the new <workflow_type>-<runtime>-queue family before redeploying the orchestrator. On the first tick after upgrade, orphan schedules on old queue names are normalized away and schedules are recreated on the new queues (no in-place rename).

Note: The SIENTIA DataOps Orchestrator is designed for production use in enterprise data environments. Ensure proper security configuration and network isolation for production deployments.