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:

### 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
```json
{
  "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

```mermaid
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
```

### Key Components

- **Scouter Workflow**: Main orchestrator for incremental data processing
- **Redis Activities**: Timestamp management for incremental processing
- **MongoDB Activities**: Raw data retrieval from collections
- **CoreScouter (Black Box)**: Complete data processing pipeline including quality gates, aggregation, and PostgreSQL export


### 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
```json
{
  "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"
    }
  }
}
```

#### Architecture

```mermaid
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]
    
    A -.-> Filters[Quality Filters]
    B -.-> Redis1[(Redis)]
    C -.-> Redis2[(Redis)] 
    D -.-> PostgreSQL[(PostgreSQL)]
    E -.-> Metrics[Prometheus]
```

### 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
```json
{
  "topic": "test_sensor_data",
  "metadata": {...},
  "num_messages": 100
}
```

#### Architecture

```mermaid
flowchart TB
    subgraph workflow [" "]
        A[1. generate_and_send_data]
    end
    
    subgraph services [" "]
        Kafka[(Kafka)]
    end
    
    A -.-> Kafka
    
    style workflow fill:none,stroke:none
    style services fill:none,stroke:none
```

## 📋 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**
   ```bash
   git clone <repository-url>
   cd sientia-dataops-scouter
   ```

2. **Create virtual environment**
   ```bash
   python3.11 -m venv venv
   source ./venv/bin/activate
   ```

3. **Install dependencies**
   ```bash
   pip install -r requirements.txt
   ```

4. **Create environment configuration file**
   ```bash
   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:
   
   ```bash
   # 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:

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

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

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

```bash
# 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
```bash
# 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

```json
{
  "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:
```bash
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.