sientia-dataops-scouter_temporal/README.md

Sientia DataOps Scouter

A high-performance, scalable data processing and ML model orchestration system built on Temporal.io for industrial data collection, processing, and analytics. The Scouter system provides enterprise-grade data ingestion from multiple sources with automatic data quality validation, aggregation, and export capabilities.

Features

Core Functionality

• Multi-Source Data Ingestion: Support for Kafka topics, direct OPC server access, and real-time triggers
• Temporal Workflow Orchestration: Robust workflow management with automatic retry policies and fault tolerance
• Data Quality Gates: Configurable filtering for null values, out-of-bounds data, and custom validation rules
• Time-Series Aggregation: Flexible aggregation functions (average, median, max, min, latest) with configurable parameters
• Multi-Database Integration: PostgreSQL for persistent storage, Redis for caching, MongoDB for data retrieval
• Real-time Monitoring: Prometheus metrics and comprehensive logging for operational visibility

Advanced Capabilities

• Incremental Data Processing: Timestamp-based data loading to avoid reprocessing
• Configurable Data Retention: Redis-based temporary storage with TTL management
• Notification System: Integrated alerting and notification management via MongoDB
• Scalable Architecture: Kubernetes-ready deployment with horizontal scaling support
• Debug Mode: Optional data package storage for debugging and troubleshooting
• Worker Autoscaling: Configurable poller behavior with aggressive autoscaling policies

Architecture

The Scouter system uses a Temporal-based workflow architecture with clear separation of concerns:

Key Components

• Worker: Main application orchestrator managing Temporal workers and task queues
• Workflows: Temporal workflow definitions for data processing orchestration
• Activities: Temporal activities implementing data processing operations
• Data Services: Database connectors and data access layer
• Quality Filters: Configurable data validation and filtering mechanisms

🔄 Workflows

The Scouter system implements a parent-child workflow pattern for data processing orchestration.

1. Scouter Workflow (scouter.py)

The Scouter workflow is the main entry point for data processing pipelines. It orchestrates the complete data ingestion process and implements a robust incremental data processing pattern.

Purpose

• Data Ingestion Orchestration: Coordinates data loading from MongoDB collections
• Timestamp Management: Tracks last processed timestamps to enable incremental processing
• Workflow Delegation: Delegates actual data processing to the CoreScouter workflow
• Data Continuity: Ensures no data is lost or reprocessed between executions

Execution Flow

1. Timestamp Retrieval: Gets the last processed timestamp from Redis for the specific workflow and schedule
2. Data Loading: Loads new data from MongoDB since the last timestamp using the collection name raw_{schedule_name}
3. Timestamp Update: Updates the last processed timestamp with the most recent data point
4. Data Processing: Delegates data processing to the CoreScouter child workflow
5. Metadata Management: Maintains workflow execution metadata throughout the process

Key Features

• Incremental Processing: Only processes new data since last execution
• Automatic Retry: Implements Temporal retry policies for fault tolerance
• Timeout Management: 60-second timeout for all activities
• Error Handling: Comprehensive error handling with notification integration

Input Parameters

{
  "topic": "sensor_data_topic",
  "schedule_name": "hourly_collection",
  "model_name": "temperature_sensors",
  "model_id": "temp_001",
  "trigger_laborious": false,
  "filters": {...},
  "schema": "sensor_data",
  "table_name": "temperature_readings",
  "retention_time": 3600,
  "model_tags": {...}
}

Architecture

flowchart LR
    A[1. get_last_data_timestamp] --> B[2. load_latest_data] --> C[3. put_last_data_timestamp] --> D[4. core_scouter 🔃]
    
    A -.-> Redis[(Redis)]
    B -.-> MongoDB[(MongoDB)]
    C -.-> Redis

2. CoreScouter Workflow (core_scouter.py)

The CoreScouter workflow implements the core data processing pipeline for industrial time-series data. It handles data quality validation, aggregation, and export operations.

Purpose

• Data Quality Validation: Applies configurable filters for data integrity
• Time-Series Aggregation: Groups and aggregates data using specified functions
• Data Organization: Groups data by tags and applies retention policies
• Persistent Storage: Exports processed data to PostgreSQL
• Metrics Collection: Records processing metrics for monitoring

Execution Flow

1. Data Quality Gate: Applies configured filters (null values, out-of-bounds, custom rules)
2. Data Aggregation: Groups data by tag and name, applies aggregation functions
3. Data Grouping: Organizes data and stores temporarily in Redis with TTL
4. Data Export: Persists processed data to PostgreSQL database
5. Metrics Recording: Writes processing metrics for operational visibility

Aggregation Functions

• lts: Latest value (most recent data point)
• avg: Average of all values in the group
• mdn: Median of all values in the group
• max: Maximum value in the group
• min: Minimum value in the group

Key Features

• Configurable Quality Gates: Multiple filter types with policy-based configuration
• Flexible Aggregation: Tag-specific aggregation function configuration
• Batch Processing: Efficient handling of large datasets
• Asynchronous Export: Non-blocking data export operations
• Comprehensive Monitoring: Detailed metrics and error reporting

Input Parameters

{
  "metadata": {...},
  "workflow_name": "scouter",
  "schedule_name": "hourly_collection",
  "model_name": "temperature_sensors",
  "model_id": "temp_001",
  "data": {...},
  "trigger_laborious": false,
  "filters": {...},
  "schema": "sensor_data",
  "table_name": "temperature_readings",
  "retention_time": 3600,
  "fill_missing_tags": false,
  "debug_data_package": false,
  "model_tags": {
    "Temperature": {
      "data_range": [-50, 150],
      "aggr_function": "avg",
      "frequency": "60000"
    }
  }
}

Additional Parameters:

• fill_missing_tags (bool): Enable filling of missing tag values with default data
• debug_data_package (bool): Store raw and processed data packages in MongoDB for debugging

Architecture

flowchart LR
    A[1. data_quality_gate] --> B[2. aggregate_data] --> C[3. group_and_hold_data] --> D[4. export_data_to_postgres] --> E[5. write_metrics]
    E --> F{debug_data_package?}
    F -->|yes| G[6. store_data_package]
    
    B -.-> Redis1[(Redis)]
    C -.-> Redis2[(Redis)] 
    D -.-> PostgreSQL[(PostgreSQL)]
    E -.-> Metrics[Prometheus]
    G -.-> MongoDB[(MongoDB)]

Debug Mode

When debug_data_package is set to true, the workflow stores both raw and processed data packages in MongoDB for debugging and troubleshooting purposes. This is useful for:

• Investigating data processing issues
• Validating data transformations
• Auditing data quality gate decisions

📋 Prerequisites

• Python 3.11+
• Temporal server/cluster
• PostgreSQL database
• Redis server
• MongoDB server
• Kafka cluster (for data ingestion)

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

2. Create virtual environment

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

3. Install dependencies

   pip install -r requirements.txt
   

4. Create environment configuration file

   cp .env.example .env
   # Edit .env with your connection details
   

5. Configure external dependencies

   You'll need to set up port forwarding or connections to external services. For example:

   # Port forwarding from Kubernetes cluster
   kubectl port-forward svc/redis-master 6379:6379
   kubectl port-forward svc/mongodb 27017:27017
   kubectl port-forward svc/kafka 9092:9092
   
   # Or connect to external services
   # Ensure services are accessible on localhost with appropriate ports
   

📦 How to Run

Running the Scouter 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 scouter 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

The script will:

• Activate the virtual environment
• Run pytest with coverage reporting
• Generate HTML coverage report
• Open the coverage report in your browser

Manual Test Execution

You can also run tests manually:

# Activate virtual environment
source ./venv/bin/activate

# Run all tests
pytest

# Run with coverage
pytest --cov=scouter --cov-report=html

# Run specific test categories
pytest tests/activities/
pytest tests/workflow/

Manual Application Execution

For manual execution without scripts:

# Activate virtual environment
source ./venv/bin/activate

# Load environment variables (if using .env file)
if [ -f .env ]; then
    export $(cat .env | grep -v '^#' | xargs)
fi

# Start the scouter worker
python -m scouter.worker.worker

Environment Configuration

Before running the application, ensure your .env file contains the necessary configuration.

🧪 Testing

Test Structure

tests/
├── activities/           # Activity implementation tests
├── workflow/            # Workflow orchestration 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=scouter --cov-report=html

# Run specific test modules
pytest tests/activities/test_redis.py
pytest tests/workflow/test_scouter.py

📊 Monitoring and Metrics

The Scouter system exposes comprehensive Prometheus metrics:

Application Metrics

• app_up: Application health status (1=healthy, 0=unhealthy)
• scouter_laborious_data_written_count: Data export operation count
• scouter_tag_changes_monitor: Tag value change monitoring

Temporal Metrics

• Workflow execution counts and durations
• Activity execution success/failure rates
• Task queue processing metrics
• Worker health and performance indicators

Database Metrics

• Connection pool utilization
• Query execution times
• Error rates and retry counts

⚙️ Configuration

Environment Variables

Variable	Description	Default	Required
TEMPORAL_HOST	Temporal server address	localhost:7233	Yes
TEMPORAL_NAMESPACE	Temporal namespace	scouter	No
POSTGRES_HOST	PostgreSQL hostname	localhost	Yes
POSTGRES_PORT	PostgreSQL port	5432	Yes
POSTGRES_USER	PostgreSQL username	sientia	Yes
POSTGRES_PASSWORD	PostgreSQL password	sientia	Yes
POSTGRES_DBNAME	PostgreSQL database	sientia	Yes
REDIS_HOST	Redis hostname	localhost	Yes
REDIS_PORT	Redis port	6379	Yes
MONGODB_URL	MongoDB connection URI	localhost:27017	Yes
KAFKA_BOOTSTRAP_SERVERS	Kafka broker addresses	localhost:9092	No
HTTP_METRICS_PORT	Prometheus metrics port	9090	No
HTTP_SDK_METRICS_PORT	Temporal SDK metrics port	9091	No
PROJECT_NAME	Project identifier for notifications	scouter	No

Worker Configuration

The worker supports advanced configuration for optimizing performance and latency:

Variable	Description	Default	Recommended
MAX_CONCURRENT_WORKFLOW_TASKS	Maximum concurrent workflow tasks	200	100-500
MAX_CONCURRENT_ACTIVITIES	Maximum concurrent activities	200	100-500
MAX_CONCURRENT_LOCAL_ACTIVITIES	Maximum concurrent local activities	200	100-500
MAX_CACHED_WORKFLOWS	Maximum cached workflow instances	200	100-500

Poller Autoscaling Configuration

The worker implements aggressive autoscaling policies for workflow and activity pollers:

Workflow Poller Behavior:

Variable	Description	Default
WORKFLOW_POLLER_BEHAVIUR_MINIMUM	Minimum workflow pollers	10
WORKFLOW_POLLER_BEHAVIUR_INITIAL	Initial workflow pollers	100
WORKFLOW_POLLER_BEHAVIUR_MAXIMUM	Maximum workflow pollers	200

Activity Poller Behavior:

Variable	Description	Default
ACTIVITY_POLLER_BEHAVIUR_MINIMUM	Minimum activity pollers	10
ACTIVITY_POLLER_BEHAVIUR_INITIAL	Initial activity pollers	100
ACTIVITY_POLLER_BEHAVIUR_MAXIMUM	Maximum activity pollers	200

Workflow Configuration

MongoDB pipeline configuration:

Scouter Workflow Input Parameters

{
  "schedule_name": "scouter-opcua-orchestrated-pipeline",
  "model_id": "1",
  "workflow_type": "scouter",
  "frequency": "5s", # Workflow execution frequency
  "max_retry_policy": 1, # Maximum number of retries for the workflow
  "read_tags": [
    {
      "tag_name": "Counter",
      "server_id": "1",
      "aggr_func": "avg",
      "tag_address": "ns=2;i=2",
      "frequency": "15000", # Tag expected frequency (used in Ingestor)
      "data_range": [
        -100,
        100
      ]
    },
    {
      "tag_name": "Rollout",
      "server_id": "1",
      "aggr_func": "mdn",
      "tag_address": "ns=2;i=3",
      "frequency": "15000",
      "data_range": [
        -100,
        100
      ]
    }
  ],
  "filters": [
    {
      "filter_name": "OUT_OF_BOUNDS_FILTER",
      "policy": "DISCARD"
    },
    {
      "filter_name": "NULL_VALUES_FILTER",
      "policy": "DISCARD"
    }
  ],
  "tag_retention_minutes": 60, # Time that tag data is cached in Redis
  "active": true,
  "updated_at": {
    "$date": "2025-08-13T18:35:01.600Z"
  }
}

🔧 Development

Project Structure

scouter/
├── activities/              # Temporal activity implementations
│   ├── activities.py       # Main activities orchestrator
│   ├── redis.py           # Redis operations (caching, timestamps)
│   ├── gates.py            # Data quality gates and filtering
│   └── mongodb.py          # MongoDB operations (data loading)
├── workflow/                # Temporal workflow definitions
│   ├── scouter.py          # Main data ingestion workflow
│   └── sub_workflows/      # Sub-workflow implementations
│       └── core_scouter.py # Core data processing workflow
├── worker/                  # Worker implementation
│   └── worker.py           # Main worker orchestrator
├── utils/                   # Utility functions
│   ├── connectors_config.py # Database configuration
│   └── quality/            # Data quality filters
├── metrics.py               # Prometheus metrics definitions
└── __init__.py

Activity Implementations

The Activities class combines multiple service classes through multiple inheritance:

• Postgres (from sientia-dataops-library): PostgreSQL data export and persistence
• Redis: Timestamp management, data caching, and temporary storage
• Gates: Data quality validation and filtering logic
• MongoDB: Data loading from raw collections

All activities support:

• Comprehensive logging and error handling
• Notification integration for errors and alerts
• Prometheus metrics collection
• Graceful shutdown and resource cleanup

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. Workflow Execution Failures

   • Review activity error logs and notifications
   • Check data quality filter configurations
   • Verify input data format and required fields

4. Performance Issues

   • Monitor Prometheus metrics for bottlenecks
   • Review database query performance
   • Check Temporal worker configuration

Debug Mode

Enable debug logging by setting the log level:

export LOG_LEVEL=DEBUG

⚡ Performance Tuning

Key Parameters

• Worker Concurrency: Adjust MAX_CONCURRENT_WORKFLOW_TASKS and MAX_CONCURRENT_ACTIVITIES (default: 200)
• Poller Autoscaling: Configure minimum, initial, and maximum poller counts for optimal throughput
• Connection Pools: Optimize database connection pool sizes (configured in build_*_config() functions)
• Data Retention: Configure Redis TTL via retention_time parameter (in seconds)
• Workflow Caching: Set MAX_CACHED_WORKFLOWS to balance memory usage and performance

Scaling Considerations

• Horizontal Scaling: Deploy multiple worker instances (each registers to scouter-queue)
• Poller Autoscaling: Workers implement aggressive autoscaling (10-200 pollers) for latency optimization
• Database Performance: Connection pooling is configured in utils/connectors_config.py
• Worker Placement: Use pod anti-affinity rules in Kubernetes for optimal distribution
• Resource Limits: Configure appropriate CPU/memory limits based on concurrency settings

🤝 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 Scouter system is designed for production use in industrial data processing environments. Ensure proper security configuration and network isolation for production deployments.