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, PI Web API endpoints, 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 with timestamp conversion
5. Metrics Recording: Writes processing metrics for operational visibility

Note: The data export step uses timestamp conversion to ensure consistent datetime formatting. The export operation receives the schema, table name, data, and timestamp conversion configuration. Conflict resolution and unique column constraints are handled by the underlying PostgreSQL activity implementation.

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

3. PI Web API Scouter Workflow (pi_web_api_scouter.py)

The PI Web API Scouter workflow serves as the entry point for PI Web API data processing pipelines. Unlike the standard Scouter workflow that loads data from MongoDB collections, this workflow directly queries PI Web API endpoints to retrieve tag values and processes them for downstream use.

Purpose

• Direct API Ingestion: Retrieves data directly from PI Web API endpoints
• Real-time Data Processing: Supports real-time and historical data retrieval
• Data Normalization: Normalizes timestamps to ensure consistency across records
• Workflow Orchestration: Delegates data processing to the CoreScouter workflow
• Error Handling: Comprehensive error handling with retry policies

Execution Flow

1. Tag Value Retrieval: Retrieves tag values from PI Web API using configured WebIds and time periods
2. Data Normalization: Normalizes timestamps to ensure all records in a batch share the same timestamp value
3. Data Validation: Validates retrieved data and handles empty responses
4. Data Processing: Delegates data processing to the CoreScouter child workflow

Note: The timestamp normalization process converts all timestamps to string format and then sets all records to the maximum timestamp value (lexicographically) found in the dataset. This ensures consistency across all records in a single batch.

Key Features

• Configurable Time Periods: Supports flexible time period configurations (e.g., '-1d', '-1h')
• Data Point Limits: Configurable maximum data points per tag via max_count parameter
• Timeout Management: Configurable API request timeouts for reliable operation
• Empty Data Handling: Gracefully handles empty responses without processing
• Standardized Processing: Uses CoreScouter for consistent data quality and export operations

Input Parameters

{
  "model_name": "pi_sensors",
  "model_id": "pi_001",
  "schedule_name": "hourly_pi_collection",
  "pi_web_api_query": {
    "endpoint": "/streamsets/recorded",
    "period": "*-1d",
    "max_count": 10,
    "api_timeout": 30
  },
  "model_tags": {
    "Temperature": {
      "webid": "F1AbCdEfGhIjKlMnOpQrStUvWxYz",
      "aggr_function": "avg",
      "data_range": [-50, 150]
    },
    "Pressure": {
      "webid": "F2AbCdEfGhIjKlMnOpQrStUvWxYz",
      "aggr_function": "max",
      "data_range": [0, 100]
    }
  },
  "trigger_laborious": false,
  "filters": {
    "OUT_OF_BOUNDS_FILTER": {"policy": "DISCARD"},
    "NULL_VALUES_FILTER": {"policy": "DISCARD"}
  },
  "schema": "sensor_data",
  "table_name": "pi_readings",
  "retention_time": 3600,
  "fill_missing_tags": false,
  "debug_data_package": false
}

PI Web API Query Parameters:

• endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded')
• period (str): Time period configuration (e.g., '-1d' for last day, '-1h' for last hour)
• max_count (int, optional): Maximum data points per tag. Defaults to 1
• api_timeout (int): Request timeout in seconds for PI Web API calls

Model Tags Configuration:

• webid (str): PI Web API WebId for the tag
• aggr_function (str): Aggregation method (avg, mdn, max, min, lts)
• data_range (list[int]): [min, max] values for data validation

Architecture

flowchart LR
    A[1. get_tag_values] --> B{data empty?}
    B -->|yes| C[Exit]
    B -->|no| D[2. core_scouter 🔃]
    
    A -.-> PI_API[(PI Web API)]
    D -.-> CoreScouter[CoreScouter Workflow]

Data Normalization

The get_tag_values activity normalizes timestamps to ensure consistency:

1. Converts all timestamps to string format using the configured datetime format
2. Identifies the maximum timestamp value (lexicographically) in the dataset
3. Sets all records to use this normalized timestamp value

This normalization ensures that all records in a single batch share the same timestamp, which is useful for batch processing and data consistency in downstream operations.

📋 Prerequisites

• Python 3.11+
• Temporal server/cluster
• PostgreSQL database
• Redis server
• MongoDB server
• Kafka cluster (for data ingestion)
• PI Web API server (for PI Web API Scouter workflow)

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
PI_WEB_API_BASE_URL	PI Web API base URL	-	Yes (for PI Web API Scouter)
PI_WEB_API_AUTH_TYPE	PI Web API authentication type ('basic' or 'bearer')	-	Yes (for PI Web API Scouter)
PI_WEB_API_AUTH_TOKEN	PI Web API authentication token	-	Yes (for PI Web API Scouter)
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_BEHAVIOUR_MINIMUM	Minimum workflow pollers	10
WORKFLOW_POLLER_BEHAVIOUR_INITIAL	Initial workflow pollers	100
WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM	Maximum workflow pollers	200

Activity Poller Behavior:

Variable	Description	Default
ACTIVITY_POLLER_BEHAVIOUR_MINIMUM	Minimum activity pollers	10
ACTIVITY_POLLER_BEHAVIOUR_INITIAL	Initial activity pollers	100
ACTIVITY_POLLER_BEHAVIOUR_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
│   ├── api.py              # PI Web API operations (tag value retrieval)
│   ├── 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
│   ├── pi_web_api_scouter.py # PI Web API 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
• API: PI Web API tag value retrieval and data normalization

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.