sientia-dataops-scouter_temporal/README.md

Sientia DataOps Scouter

A high-performance, scalable data processing 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
• Test Data Generation: Built-in fake data generation for development and testing

Architecture

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

System Architecture Diagram

graph TB
    %% External Data Sources
    Kafka[Kafka Topics]
    OPC[OPC UA Servers]
    
    %% Temporal Infrastructure
    subgraph "Temporal Cluster"
        TC[Temporal Client]
        TW[Temporal Workers]
        TQ1[scouter-queue]
        TQ2[fake_data-queue]
    end
    
    %% Scouter Application
    subgraph "Scouter Application"
        Worker[Worker Process]
        
        %% Workflows
        subgraph "Workflows"
            ScouterWF[Scouter Workflow<br/>Data Orchestration]
            CoreWF[CoreScouter Workflow<br/>Data Processing]
            FakeWF[FakeData Workflow<br/>Test Data Generation]
        end
        
        %% Activities Layer
        subgraph "Activities Layer"
            ActMain[Activities<br/>Unified Interface]
            RedisAct[Redis Activities<br/>Caching & Timestamps]
            MongoAct[MongoDB Activities<br/>Data Retrieval]
            PostgresAct[PostgreSQL Activities<br/>Data Persistence]
            GatesAct[Quality Gates<br/>Data Validation]
            FakerAct[Faker Activities<br/>Test Data Generation]
        end
    end
    
    %% Data Storage Layer
    subgraph "Data Storage"
        MongoDB[(MongoDB<br/>Raw Data Collections)]
        Redis[(Redis<br/>Cache & Timestamps)]
        PostgreSQL[(PostgreSQL<br/>Processed Data)]
    end
    
    %% Monitoring
    subgraph "Monitoring & Observability"
        Prometheus[Prometheus Metrics]
        Logs[Application Logs]
        Notifications[Notification System]
    end
    
    %% Data Flow Connections
    Kafka -->|Raw Data| MongoDB
    OPC -->|Sensor Data| MongoDB
    
    %% Worker to Temporal
    Worker --> TC
    TC --> TQ1
    TC --> TQ2
    TQ1 --> ScouterWF
    TQ2 --> FakeWF
    
    %% Workflow Orchestration
    ScouterWF -->|Child Workflow| CoreWF
    ScouterWF --> RedisAct
    ScouterWF --> MongoAct
    
    %% Core Processing Flow
    CoreWF --> GatesAct
    CoreWF --> RedisAct
    CoreWF --> PostgresAct
    
    %% Test Data Flow
    FakeWF --> FakerAct
    FakerAct -->|Generate| Kafka
    
    %% Activities to Data Sources
    ActMain --> RedisAct
    ActMain --> MongoAct
    ActMain --> PostgresAct
    ActMain --> GatesAct
    
    RedisAct <--> Redis
    MongoAct --> MongoDB
    PostgresAct --> PostgreSQL
    
    %% Monitoring Connections
    Worker --> Prometheus
    ActMain --> Logs
    ActMain --> Notifications
    
    %% Styling
    classDef workflow fill:#e1f5fe,stroke:#01579b,stroke-width:2px
    classDef activity fill:#f3e5f5,stroke:#4a148c,stroke-width:2px
    classDef storage fill:#e8f5e8,stroke:#1b5e20,stroke-width:2px
    classDef external fill:#fff3e0,stroke:#e65100,stroke-width:2px
    classDef monitoring fill:#fce4ec,stroke:#880e4f,stroke-width:2px
    
    class ScouterWF,CoreWF,FakeWF workflow
    class ActMain,RedisAct,MongoAct,PostgresAct,GatesAct,FakerAct activity
    class MongoDB,Redis,PostgreSQL storage
    class Kafka,OPC external
    class Prometheus,Logs,Notifications monitoring

Data Processing Flow

sequenceDiagram
    participant K as Kafka/OPC
    participant M as MongoDB
    participant SW as Scouter Workflow
    participant R as Redis
    participant CW as CoreScouter Workflow
    participant G as Quality Gates
    participant P as PostgreSQL
    
    %% Data Ingestion
    K->>M: Raw sensor data
    
    %% Workflow Execution
    SW->>R: Get last timestamp
    SW->>M: Load new data since timestamp
    SW->>R: Update last timestamp
    SW->>CW: Process data (child workflow)
    
    %% Core Processing
    CW->>G: Apply quality filters
    G->>G: Validate data ranges
    G->>G: Remove null values
    CW->>G: Aggregate data by tags
    CW->>R: Store grouped data (TTL)
    CW->>P: Export to PostgreSQL
    
    %% Monitoring
    CW->>CW: Record metrics
    P-->>SW: Processing complete

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

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

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,
  "model_tags": {
    "Temperature": {
      "data_range": [-50, 150],
      "aggr_function": "avg",
      "frequency": "60000"
    }
  }
}

3. FakeData Workflow (fake_data.py)

The FakeData workflow generates synthetic industrial sensor data for testing and development purposes. It's designed to simulate realistic data flows without requiring actual industrial data sources.

Purpose

• Test Data Generation: Creates realistic sensor data for development and testing
• Pipeline Validation: Tests data processing workflows with known data
• Load Testing: Generates configurable data volumes for performance testing
• Demonstration: Shows data flow patterns and processing capabilities

Execution Flow

1. Data Generation: Creates synthetic sensor readings with realistic values
2. Kafka Publishing: Sends generated data to specified Kafka topics
3. Quality Assurance: Ensures data format consistency and completeness
4. Monitoring: Tracks generation and publishing metrics

Key Features

• Realistic Data: Generates data within realistic industrial ranges
• Configurable Volume: Adjustable message counts for different testing scenarios
• Random Variation: Includes realistic data variations and occasional null values
• Kafka Integration: Direct integration with Kafka for data streaming
• Error Handling: Comprehensive error handling and logging

Input Parameters

{
  "topic": "test_sensor_data",
  "metadata": {...},
  "num_messages": 100
}

📋 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

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
│   ├── gates.py            # Data quality gates
│   ├── mongodb.py          # MongoDB operations
│   └── faker.py            # Test data generation
├── workflow/                # Temporal workflow definitions
│   ├── scouter.py          # Main data ingestion workflow
│   ├── fake_data.py        # Test data 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

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
• Connection Pools: Optimize database connection pool sizes
• Data Retention: Configure Redis TTL based on processing requirements
• Batch Sizes: Adjust data processing batch sizes for optimal throughput

Scaling Considerations

• Horizontal Scaling: Deploy multiple worker instances
• Task Queue Distribution: Use multiple task queues for different workflow types
• Database Performance: Optimize indexes and connection pooling
• Kafka Partitioning: Configure appropriate partition counts for data ingestion

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